[KIP-848] Added support for testing with new 'consumer' group protocol. (#1812)

Author: pranavrth

.semaphore/semaphore.yml

@@ -186,12 +186,20 @@ blocks:
         commands:
           - '[[ -z $DOCKERHUB_APIKEY ]] || docker login --username $DOCKERHUB_USER --password $DOCKERHUB_APIKEY'
       jobs:
-        - name: Build
+        - name: Build and Tests with 'classic' group protocol
+          commands:
+            - sem-version python 3.9
+            # use a virtualenv
+            - python3 -m venv _venv && source _venv/bin/activate
+            - chmod u+r+x tools/source-package-verification.sh
+            - tools/source-package-verification.sh
+        - name: Build and Tests with 'consumer' group protocol
           commands:
-            - sem-version python 3.8
+            - sem-version python 3.9
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
+            - export TEST_CONSUMER_GROUP_PROTOCOL=consumer
             - tools/source-package-verification.sh
   - name: "Source package verification with Python 3 (Linux arm64)"
     dependencies: []
@@ -207,7 +215,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.8
+            - sem-version python 3.9
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -226,7 +234,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.8
+            - sem-version python 3.9
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -245,7 +253,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.8
+            - sem-version python 3.9
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh

requirements/requirements-tests-install.txt

@@ -2,4 +2,5 @@
 -r requirements-schemaregistry.txt  
 -r requirements-avro.txt  
 -r requirements-protobuf.txt  
--r requirements-json.txt 
+-r requirements-json.txt
+tests/trivup/trivup-0.12.7.tar.gz

requirements/requirements-tests.txt

@@ -4,5 +4,4 @@ urllib3 >= 2.0.0,<3; python_version > "3.7"
 flake8
 pytest
 pytest-timeout
-requests-mock
-trivup>=0.8.3
+requests-mock

tests/README.md

@@ -18,7 +18,8 @@ A python3 env suitable for running tests:
 
     $ python3 -m venv venv_test
     $ source venv_test/bin/activate
-    $ python3 -m pip install .[tests]
+    $ python3 -m pip install -r requirements/requirements-tests-install.txt
+    $ python3 -m pip install .
 
 When you're finished with it:
 

tests/common/__init__.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2024 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+from confluent_kafka import Consumer, DeserializingConsumer
+from confluent_kafka.avro import AvroConsumer
+
+_GROUP_PROTOCOL_ENV = 'TEST_CONSUMER_GROUP_PROTOCOL'
+_TRIVUP_CLUSTER_TYPE_ENV = 'TEST_TRIVUP_CLUSTER_TYPE'
+
+
+def _update_conf_group_protocol(conf=None):
+    if conf is not None and 'group.id' in conf and TestUtils.use_group_protocol_consumer():
+        conf['group.protocol'] = 'consumer'
+
+
+def _trivup_cluster_type_kraft():
+    return _TRIVUP_CLUSTER_TYPE_ENV in os.environ and os.environ[_TRIVUP_CLUSTER_TYPE_ENV] == 'kraft'
+
+
+class TestUtils:
+    @staticmethod
+    def use_kraft():
+        return TestUtils.use_group_protocol_consumer() or _trivup_cluster_type_kraft()
+
+    @staticmethod
+    def use_group_protocol_consumer():
+        return _GROUP_PROTOCOL_ENV in os.environ and os.environ[_GROUP_PROTOCOL_ENV] == 'consumer'
+
+
+class TestConsumer(Consumer):
+    def __init__(self, conf=None, **kwargs):
+        _update_conf_group_protocol(conf)
+        super(TestConsumer, self).__init__(conf, **kwargs)
+
+
+class TestDeserializingConsumer(DeserializingConsumer):
+    def __init__(self, conf=None, **kwargs):
+        _update_conf_group_protocol(conf)
+        super(TestDeserializingConsumer, self).__init__(conf, **kwargs)
+
+
+class TestAvroConsumer(AvroConsumer):
+    def __init__(self, conf=None, **kwargs):
+        _update_conf_group_protocol(conf)
+        super(TestAvroConsumer, self).__init__(conf, **kwargs)

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)
@@ -58,6 +58,8 @@ def verify_admin_acls(admin_client,
     for acl_binding, f in fs.items():
         f.result()  # trigger exception if there was an error
 
+    time.sleep(1)
+
     acl_binding_filter1 = AclBindingFilter(ResourceType.ANY, None, ResourcePatternType.ANY,
                                            None, None, AclOperation.ANY, AclPermissionType.ANY)
     acl_binding_filter2 = AclBindingFilter(ResourceType.ANY, None, ResourcePatternType.PREFIXED,
@@ -83,6 +85,8 @@ def verify_admin_acls(admin_client,
         "Deleted ACL bindings don't match, actual {} expected {}".format(deleted_acl_bindings,
                                                                          expected_acl_bindings)
 
+    time.sleep(1)
+
     #
     # Delete the ACLs with TOPIC and GROUP
     #
@@ -94,6 +98,9 @@ def verify_admin_acls(admin_client,
         assert deleted_acl_bindings == expected, \
             "Deleted ACL bindings don't match, actual {} expected {}".format(deleted_acl_bindings,
                                                                              expected)
+
+    time.sleep(1)
+
     #
     # All the ACLs should have been deleted
     #
@@ -201,14 +208,14 @@ def test_basic_operations(kafka_cluster):
     # Second iteration: create topic.
     #
     for validate in (True, False):
-        our_topic = kafka_cluster.create_topic(topic_prefix,
-                                               {
-                                                   "num_partitions": num_partitions,
-                                                   "config": topic_config,
-                                                   "replication_factor": 1,
-                                               },
-                                               validate_only=validate
-                                               )
+        our_topic = kafka_cluster.create_topic_and_wait_propogation(topic_prefix,
+                                                                    {
+                                                                        "num_partitions": num_partitions,
+                                                                        "config": topic_config,
+                                                                        "replication_factor": 1,
+                                                                    },
+                                                                    validate_only=validate
+                                                                    )
 
     admin_client = kafka_cluster.admin()
 
@@ -270,7 +277,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
@@ -345,6 +352,8 @@ def verify_config(expconfig, configs):
     fs = admin_client.alter_configs([resource])
     fs[resource].result()  # will raise exception on failure
 
+    time.sleep(1)
+
     #
     # Read the config back again and verify.
     #

tests/integration/admin/test_delete_records.py

@@ -25,11 +25,11 @@ def test_delete_records(kafka_cluster):
     admin_client = kafka_cluster.admin()
 
     # Create a topic with a single partition
-    topic = kafka_cluster.create_topic("test-del-records",
-                                       {
-                                           "num_partitions": 1,
-                                           "replication_factor": 1,
-                                       })
+    topic = kafka_cluster.create_topic_and_wait_propogation("test-del-records",
+                                                            {
+                                                                "num_partitions": 1,
+                                                                "replication_factor": 1,
+                                                            })
 
     # Create Producer instance
     p = kafka_cluster.producer()
@@ -73,16 +73,17 @@ def test_delete_records_multiple_topics_and_partitions(kafka_cluster):
     admin_client = kafka_cluster.admin()
     num_partitions = 3
     # Create two topics with a single partition
-    topic = kafka_cluster.create_topic("test-del-records",
-                                       {
-                                           "num_partitions": num_partitions,
-                                           "replication_factor": 1,
-                                       })
-    topic2 = kafka_cluster.create_topic("test-del-records2",
-                                        {
-                                           "num_partitions": num_partitions,
-                                           "replication_factor": 1,
-                                        })
+    topic = kafka_cluster.create_topic_and_wait_propogation("test-del-records",
+                                                            {
+                                                                "num_partitions": num_partitions,
+                                                                "replication_factor": 1,
+                                                            })
+    topic2 = kafka_cluster.create_topic_and_wait_propogation("test-del-records2",
+                                                             {
+                                                                 "num_partitions": num_partitions,
+                                                                 "replication_factor": 1,
+                                                             })
+
     topics = [topic, topic2]
     partitions = list(range(num_partitions))
     # Create Producer instance

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
 
 
@@ -196,20 +204,24 @@ def test_describe_operations(sasl_cluster):
 
     # Create Topic
     topic_config = {"compression.type": "gzip"}
-    our_topic = sasl_cluster.create_topic(topic_prefix,
-                                          {
-                                              "num_partitions": 1,
-                                              "config": topic_config,
-                                              "replication_factor": 1,
-                                          },
-                                          validate_only=False
-                                          )
+    our_topic = sasl_cluster.create_topic_and_wait_propogation(topic_prefix,
+                                                               {
+                                                                   "num_partitions": 1,
+                                                                   "config": topic_config,
+                                                                   "replication_factor": 1,
+                                                               },
+                                                               validate_only=False
+                                                               )
 
     # 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)