PR comments

Author: pranavrth

tests/integration/admin/test_basic_operations.py

@@ -13,10 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import confluent_kafka
 import struct
 import time
-from confluent_kafka import ConsumerGroupTopicPartitions, TopicPartition, ConsumerGroupState
+
+from confluent_kafka import ConsumerGroupTopicPartitions, TopicPartition, ConsumerGroupState, KafkaError
 from confluent_kafka.admin import (NewPartitions, ConfigResource,
                                    AclBinding, AclBindingFilter, ResourceType,
                                    ResourcePatternType, AclOperation, AclPermissionType)
@@ -55,6 +55,7 @@ def verify_admin_acls(admin_client,
                                "User:test-user-2", "*", AclOperation.ALL, AclPermissionType.ALLOW)
 
     fs = admin_client.create_acls([acl_binding_1, acl_binding_2, acl_binding_3])
+    time.sleep(1)
     for acl_binding, f in fs.items():
         f.result()  # trigger exception if there was an error
 
@@ -78,6 +79,7 @@ def verify_admin_acls(admin_client,
     #
     expected_acl_bindings = [acl_binding_2, acl_binding_3]
     fs = admin_client.delete_acls([acl_binding_filter2])
+    time.sleep(1)
     deleted_acl_bindings = sorted(fs[acl_binding_filter2].result())
     assert deleted_acl_bindings == expected_acl_bindings, \
         "Deleted ACL bindings don't match, actual {} expected {}".format(deleted_acl_bindings,
@@ -89,6 +91,7 @@ def verify_admin_acls(admin_client,
     expected_acl_bindings = [[acl_binding_1], []]
     delete_acl_binding_filters = [acl_binding_filter3, acl_binding_filter4]
     fs = admin_client.delete_acls(delete_acl_binding_filters)
+    time.sleep(1)
     for acl_binding, expected in zip(delete_acl_binding_filters, expected_acl_bindings):
         deleted_acl_bindings = sorted(fs[acl_binding].result())
         assert deleted_acl_bindings == expected, \
@@ -209,6 +212,7 @@ def test_basic_operations(kafka_cluster):
                                                },
                                                validate_only=validate
                                                )
+        time.sleep(1)
 
     admin_client = kafka_cluster.admin()
 
@@ -270,7 +274,7 @@ def consume_messages(group_id, num_messages=None):
                     print('Read all the required messages: exiting')
                     break
             except ConsumeError as e:
-                if msg is not None and e.code == confluent_kafka.KafkaError._PARTITION_EOF:
+                if msg is not None and e.code == KafkaError._PARTITION_EOF:
                     print('Reached end of %s [%d] at offset %d' % (
                           msg.topic(), msg.partition(), msg.offset()))
                     eof_reached[(msg.topic(), msg.partition())] = True
@@ -343,6 +347,7 @@ def verify_config(expconfig, configs):
         resource.set_config(key, value)
 
     fs = admin_client.alter_configs([resource])
+    time.sleep(1)
     fs[resource].result()  # will raise exception on failure
 
     #

tests/integration/admin/test_describe_operations.py

@@ -13,12 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
 import pytest
+
 from confluent_kafka.admin import (AclBinding, AclBindingFilter, ResourceType,
                                    ResourcePatternType, AclOperation, AclPermissionType)
 from confluent_kafka.error import ConsumeError
 from confluent_kafka import ConsumerGroupState, TopicCollection
 
+from tests.common import TestUtils
+
 topic_prefix = "test-topic"
 
 
@@ -82,10 +86,12 @@ def perform_admin_operation_sync(operation, *arg, **kwargs):
 
 def create_acls(admin_client, acl_bindings):
     perform_admin_operation_sync(admin_client.create_acls, acl_bindings)
+    time.sleep(1)
 
 
 def delete_acls(admin_client, acl_binding_filters):
     perform_admin_operation_sync(admin_client.delete_acls, acl_binding_filters)
+    time.sleep(1)
 
 
 def verify_provided_describe_for_authorized_operations(
@@ -115,6 +121,7 @@ def verify_provided_describe_for_authorized_operations(
     acl_binding = AclBinding(restype, resname, ResourcePatternType.LITERAL,
                              "User:sasl_user", "*", operation_to_allow, AclPermissionType.ALLOW)
     create_acls(admin_client, [acl_binding])
+    time.sleep(1)
 
     # Check with updated authorized operations
     desc = perform_admin_operation_sync(describe_fn, *arg, **kwargs)
@@ -126,6 +133,7 @@ def verify_provided_describe_for_authorized_operations(
     acl_binding_filter = AclBindingFilter(restype, resname, ResourcePatternType.ANY,
                                           None, None, AclOperation.ANY, AclPermissionType.ANY)
     delete_acls(admin_client, [acl_binding_filter])
+    time.sleep(1)
     return desc
 
 
@@ -204,12 +212,17 @@ def test_describe_operations(sasl_cluster):
                                           },
                                           validate_only=False
                                           )
+    time.sleep(1)
 
     # Verify Authorized Operations in Describe Topics
     verify_describe_topics(admin_client, our_topic)
 
     # Verify Authorized Operations in Describe Groups
-    verify_describe_groups(sasl_cluster, admin_client, our_topic)
+    # Skip this test if using group protocol `consumer`
+    # as there is new RPC for describe_groups() in
+    # group protocol `consumer` case.
+    if not TestUtils.use_group_protocol_consumer():
+        verify_describe_groups(sasl_cluster, admin_client, our_topic)
 
     # Delete Topic
     perform_admin_operation_sync(admin_client.delete_topics, [our_topic], operation_timeout=0, request_timeout=10)

tests/integration/admin/test_incremental_alter_configs.py

@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 from confluent_kafka.admin import ConfigResource, \
     ConfigEntry, ResourceType, \
     AlterConfigOpType
@@ -58,12 +60,14 @@ def test_incremental_alter_configs(kafka_cluster):
                                                "config": topic_config,
                                                "replication_factor": 1,
                                            })
+    time.sleep(1)
     our_topic2 = kafka_cluster.create_topic(topic_prefix2,
                                             {
                                                 "num_partitions": num_partitions,
                                                 "config": topic_config,
                                                 "replication_factor": 1,
                                             })
+    time.sleep(1)
 
     admin_client = kafka_cluster.admin()
 
@@ -100,6 +104,7 @@ def test_incremental_alter_configs(kafka_cluster):
     # Incrementally alter some configuration values
     #
     fs = admin_client.incremental_alter_configs([res1, res2])
+    time.sleep(1)
 
     assert_operation_succeeded(fs, 2)
 
@@ -131,6 +136,7 @@ def test_incremental_alter_configs(kafka_cluster):
     # Incrementally alter some configuration values
     #
     fs = admin_client.incremental_alter_configs([res2])
+    time.sleep(1)
 
     assert_operation_succeeded(fs, 1)
 

tests/integration/admin/test_list_offsets.py

@@ -13,8 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from confluent_kafka.admin import ListOffsetsResultInfo, OffsetSpec
+import time
+
 from confluent_kafka import TopicPartition, IsolationLevel
+from confluent_kafka.admin import ListOffsetsResultInfo, OffsetSpec
 
 
 def test_list_offsets(kafka_cluster):
@@ -32,6 +34,7 @@ def test_list_offsets(kafka_cluster):
                                            "num_partitions": 1,
                                            "replication_factor": 1,
                                        })
+    time.sleep(1)
 
     # Create Producer instance
     p = kafka_cluster.producer()

tests/integration/admin/test_user_scram_credentials.py

@@ -13,13 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import pytest
+import time
 import concurrent
+import pytest
+
 from confluent_kafka.admin import UserScramCredentialsDescription, UserScramCredentialUpsertion, \
                                   UserScramCredentialDeletion, ScramCredentialInfo, \
                                   ScramMechanism
 from confluent_kafka.error import KafkaException, KafkaError
 
+from tests.common import TestUtils
 
 def test_user_scram_credentials(kafka_cluster):
     """
@@ -47,22 +50,28 @@ def test_user_scram_credentials(kafka_cluster):
     futmap = admin_client.alter_user_scram_credentials([UserScramCredentialUpsertion(newuser,
                                                         ScramCredentialInfo(mechanism, iterations),
                                                         password, salt)])
+    time.sleep(1)
     fut = futmap[newuser]
     result = fut.result()
     assert result is None
 
     # Try upsertion for newuser,SCRAM_SHA_256 and add newuser,SCRAM_SHA_512
-    futmap = admin_client.alter_user_scram_credentials([UserScramCredentialUpsertion(
-                                                            newuser,
-                                                            ScramCredentialInfo(
-                                                                mechanism, iterations),
-                                                            password, salt),
-                                                        UserScramCredentialUpsertion(
-                                                            newuser,
-                                                            ScramCredentialInfo(
-                                                                ScramMechanism.SCRAM_SHA_512, 10000),
-                                                            password)
-                                                        ])
+    request = [UserScramCredentialUpsertion(newuser,
+                                            ScramCredentialInfo(
+                                                mechanism, iterations),
+                                            password, salt),
+               UserScramCredentialUpsertion(newuser,
+                                            ScramCredentialInfo(
+                                                ScramMechanism.SCRAM_SHA_512, 10000),
+                                            password)]
+
+    if TestUtils.use_kraft():
+        futmap = admin_client.alter_user_scram_credentials([request[0]])
+        time.sleep(1)
+        futmap = admin_client.alter_user_scram_credentials([request[1]])
+    else:
+        futmap = admin_client.alter_user_scram_credentials(request)
+    time.sleep(1)
     fut = futmap[newuser]
     result = fut.result()
     assert result is None
@@ -72,6 +81,7 @@ def test_user_scram_credentials(kafka_cluster):
                                                             newuser,
                                                             ScramMechanism.SCRAM_SHA_512)
                                                         ])
+    time.sleep(1)
     fut = futmap[newuser]
     result = fut.result()
     assert result is None
@@ -101,6 +111,7 @@ def test_user_scram_credentials(kafka_cluster):
 
     # Delete newuser
     futmap = admin_client.alter_user_scram_credentials([UserScramCredentialDeletion(newuser, mechanism)])
+    time.sleep(1)
     assert isinstance(futmap, dict)
     assert len(futmap) == 1
     assert newuser in futmap

tests/integration/cluster_fixture.py

@@ -23,7 +23,8 @@
 from confluent_kafka import Producer, SerializingProducer
 from confluent_kafka.admin import AdminClient, NewTopic
 from confluent_kafka.schema_registry.schema_registry_client import SchemaRegistryClient
-from ..common import TestDeserializingConsumer, TestConsumer
+
+from tests.common import TestDeserializingConsumer, TestConsumer
 
 
 class KafkaClusterFixture(object):

tests/integration/conftest.py

@@ -17,9 +17,9 @@
 #
 
 import os
-from ..common import TestUtils
 import pytest
 
+from tests.common import TestUtils
 from tests.integration.cluster_fixture import TrivupFixture
 from tests.integration.cluster_fixture import ByoFixture
 
@@ -34,20 +34,24 @@ def _broker_conf():
     return broker_conf
 
 
+def _broker_version():
+    return 'trunk@f6c9feea76d01a46319b0ca602d70aa855057b07' if TestUtils.use_group_protocol_consumer() else '3.7.0'
+
+
 def create_trivup_cluster(conf={}):
     trivup_fixture_conf = {'with_sr': True,
                            'debug': True,
                            'cp_version': '7.6.0',
                            'kraft': TestUtils.use_kraft(),
-                           'version': 'trunk@f6c9feea76d01a46319b0ca602d70aa855057b07',
+                           'version': _broker_version(),
                            'broker_conf': _broker_conf()}
     trivup_fixture_conf.update(conf)
     return TrivupFixture(trivup_fixture_conf)
 
 
 def create_sasl_cluster(conf={}):
     trivup_fixture_conf = {'with_sr': False,
-                           'version': 'trunk@f6c9feea76d01a46319b0ca602d70aa855057b07',
+                           'version': _broker_version(),
                            'sasl_mechanism': "PLAIN",
                            'kraft': TestUtils.use_kraft(),
                            'sasl_users': 'sasl_user=sasl_user',

tests/integration/integration_test.py

@@ -20,7 +20,6 @@
 
 """ Test script for confluent_kafka module """
 
-import confluent_kafka
 import os
 import time
 import uuid
@@ -29,7 +28,10 @@
 import gc
 import struct
 import re
-from ..common import TestConsumer, TestAvroConsumer
+
+import confluent_kafka
+
+from tests.common import TestConsumer, TestAvroConsumer
 
 try:
     # Memory tracker

tests/integration/producer/test_transactions.py

@@ -18,10 +18,10 @@
 import inspect
 import sys
 from uuid import uuid1
-from ...common import TestConsumer
 
 from confluent_kafka import KafkaError
 
+from tests.common import TestConsumer
 
 def called_by():
     if sys.version_info < (3, 5):

tests/test_Consumer.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python
 
+import pytest
+
 from confluent_kafka import (Consumer, TopicPartition, KafkaError,
                              KafkaException, TIMESTAMP_NOT_AVAILABLE,
                              OFFSET_INVALID, libversion)
-import pytest
-from .common import TestConsumer
+
+from tests.common import TestConsumer
 
 
 def test_basic_api():

tests/test_Producer.py

@@ -1,11 +1,12 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 import pytest
+from struct import pack
 
 from confluent_kafka import Producer, KafkaError, KafkaException, \
     TopicPartition, libversion
-from struct import pack
-from .common import TestConsumer
+
+from tests.common import TestConsumer
 
 
 def error_cb(err):

tests/test_log.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python
 
 from io import StringIO
+import logging
+
 import confluent_kafka
 import confluent_kafka.avro
-import logging
-from .common import TestConsumer, TestAvroConsumer
+
+from tests.common import TestConsumer, TestAvroConsumer
 
 
 class CountingFilter(logging.Filter):