[KIP-848] Test changes required for KIP-848 Preview (#1967)

* Fixed test_basic_operations integration test failing intermittently.

* Upgraded trivup to v0.12.10

* 1) Skipping few tests related to session.timeout.ms for the 'consumer' protocol
2) Fixed test related to assign and unassign in incremental assignor case for 'consumer' protocol

* Updated member id test for KIP-1082

* Skipping consumer error tests for 'consumer' protocol for now as it uses session.timeout.ms

* Using librdkafka v2.10.0-RC3. Using Java 17 for consumer protocol integration tests.

* Disabled test causing MacOS fatal error issue

Author: pranavrth

.semaphore/semaphore.yml

@@ -8,7 +8,7 @@ execution_time_limit:
 global_job_config:
   env_vars:
     - name: LIBRDKAFKA_VERSION
-      value: v2.8.0
+      value: v2.10.0-RC3
   prologue:
     commands:
       - checkout
@@ -208,6 +208,7 @@ blocks:
         - name: Build and Tests with 'consumer' group protocol
           commands:
             - sem-version python 3.9
+            - sem-version java 17
             # 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

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

tests/common/__init__.py

@@ -28,9 +28,26 @@ def _trivup_cluster_type_kraft():
 
 
 class TestUtils:
+    @staticmethod
+    def broker_version():
+        return '4.0.0' if TestUtils.use_group_protocol_consumer() else '3.9.0'
+
+    @staticmethod
+    def broker_conf():
+        broker_conf = ['transaction.state.log.replication.factor=1',
+                       'transaction.state.log.min.isr=1']
+        if TestUtils.use_group_protocol_consumer():
+            broker_conf.append('group.coordinator.rebalance.protocols=classic,consumer')
+        return broker_conf
+
+    @staticmethod
+    def _broker_major_version():
+        return int(TestUtils.broker_version().split('.')[0])
+
     @staticmethod
     def use_kraft():
-        return TestUtils.use_group_protocol_consumer() or _trivup_cluster_type_kraft()
+        return (TestUtils.use_group_protocol_consumer() or
+                _trivup_cluster_type_kraft())
 
     @staticmethod
     def use_group_protocol_consumer():
@@ -41,8 +58,35 @@ 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'
 
+    @staticmethod
+    def remove_forbidden_conf_group_protocol_consumer(conf):
+        if conf is None:
+            return
+        if TestUtils.use_group_protocol_consumer():
+            forbidden_conf_properties = ["session.timeout.ms",
+                                         "partition.assignment.strategy",
+                                         "heartbeat.interval.ms",
+                                         "group.protocol.type"]
+            for prop in forbidden_conf_properties:
+                if prop in conf:
+                    print("Skipping setting forbidden configuration {prop} for `CONSUMER` protocol")
+                    del conf[prop]
+
 
 class TestConsumer(Consumer):
     def __init__(self, conf=None, **kwargs):
         TestUtils.update_conf_group_protocol(conf)
+        TestUtils.remove_forbidden_conf_group_protocol_consumer(conf)
         super(TestConsumer, self).__init__(conf, **kwargs)
+
+    def assign(self, partitions):
+        if TestUtils.use_group_protocol_consumer():
+            super(TestConsumer, self).incremental_assign(partitions)
+        else:
+            super(TestConsumer, self).assign(partitions)
+
+    def unassign(self, partitions):
+        if TestUtils.use_group_protocol_consumer():
+            super(TestConsumer, self).incremental_unassign(partitions)
+        else:
+            super(TestConsumer, self).unassign()

tests/common/schema_registry/__init__.py

@@ -24,10 +24,12 @@
 class TestDeserializingConsumer(DeserializingConsumer):
     def __init__(self, conf=None, **kwargs):
         TestUtils.update_conf_group_protocol(conf)
+        TestUtils.remove_forbidden_conf_group_protocol_consumer(conf)
         super(TestDeserializingConsumer, self).__init__(conf, **kwargs)
 
 
 class TestAvroConsumer(AvroConsumer):
     def __init__(self, conf=None, **kwargs):
         TestUtils.update_conf_group_protocol(conf)
+        TestUtils.remove_forbidden_conf_group_protocol_consumer(conf)
         super(TestAvroConsumer, self).__init__(conf, **kwargs)

tests/integration/admin/test_basic_operations.py

@@ -277,7 +277,8 @@ 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 == KafkaError._PARTITION_EOF:
+                if e.code == KafkaError._PARTITION_EOF:
+                    msg = e.kafka_message
                     print('Reached end of %s [%d] at offset %d' % (
                           msg.topic(), msg.partition(), msg.offset()))
                     eof_reached[(msg.topic(), msg.partition())] = True

tests/integration/conftest.py

@@ -26,38 +26,26 @@
 work_dir = os.path.dirname(os.path.realpath(__file__))
 
 
-def _broker_conf():
-    broker_conf = ['transaction.state.log.replication.factor=1',
-                   'transaction.state.log.min.isr=1']
-    if TestUtils.use_group_protocol_consumer():
-        broker_conf.append('group.coordinator.rebalance.protocols=classic,consumer')
-    return broker_conf
-
-
-def _broker_version():
-    return 'trunk@3a0efa2845e6a0d237772adfe6364579af50ce18' if TestUtils.use_group_protocol_consumer() else '3.8.0'
-
-
 def create_trivup_cluster(conf={}):
     trivup_fixture_conf = {'with_sr': True,
                            'debug': True,
                            'cp_version': '7.6.0',
                            'kraft': TestUtils.use_kraft(),
-                           'version': _broker_version(),
-                           'broker_conf': _broker_conf()}
+                           'version': TestUtils.broker_version(),
+                           'broker_conf': TestUtils.broker_conf()}
     trivup_fixture_conf.update(conf)
     return TrivupFixture(trivup_fixture_conf)
 
 
 def create_sasl_cluster(conf={}):
     trivup_fixture_conf = {'with_sr': False,
-                           'version': _broker_version(),
+                           'version': TestUtils.broker_version(),
                            'sasl_mechanism': "PLAIN",
                            'kraft': TestUtils.use_kraft(),
                            'sasl_users': 'sasl_user=sasl_user',
                            'debug': True,
                            'cp_version': 'latest',
-                           'broker_conf': _broker_conf()}
+                           'broker_conf': TestUtils.broker_conf()}
     trivup_fixture_conf.update(conf)
     return TrivupFixture(trivup_fixture_conf)
 

tests/integration/consumer/test_consumer_error.py

@@ -21,6 +21,7 @@
 
 from confluent_kafka.error import ConsumeError
 from confluent_kafka.serialization import StringSerializer
+from tests.common import TestUtils
 
 
 def test_consume_error(kafka_cluster):
@@ -46,6 +47,14 @@ def test_consume_error(kafka_cluster):
         "Expected _PARTITION_EOF, not {}".format(exc_info)
 
 
+# Skipping the test for consumer protocol for now. Update the test to use
+# IncrementalAlterConfigs Admin operation to update
+# group.session.timeout.ms and enable the test again.
+@pytest.mark.skipif(TestUtils.use_group_protocol_consumer(),
+                    reason="session.timeout.ms is not supported on client side for "
+                    "consumer protocol. Update this test to use IncrementalAlterConfigs "
+                    "Admin operation to update group.session.timeout.ms and enable "
+                    "the test again.")
 def test_consume_error_commit(kafka_cluster):
     """
     Tests to ensure that we handle messages with errors when commiting.
@@ -63,13 +72,21 @@ def test_consume_error_commit(kafka_cluster):
     try:
         # Since the session timeout value is low, JoinGroupRequest will fail
         # and we get error in a message while polling.
-        m = consumer.poll(1)
+        m = consumer.poll(2)
         consumer.commit(m)
     except KafkaException as e:
         assert e.args[0].code() == KafkaError._INVALID_ARG, \
             "Expected INVALID_ARG, not {}".format(e)
 
 
+# Skipping the test for consumer protocol for now. Update the test to use
+# IncrementalAlterConfigs Admin operation to update
+# group.session.timeout.ms and enable the test again.
+@pytest.mark.skipif(TestUtils.use_group_protocol_consumer(),
+                    reason="session.timeout.ms is not supported on client side for "
+                    "consumer protocol. Update this test to use IncrementalAlterConfigs "
+                    "Admin operation to update group.session.timeout.ms and enable "
+                    "the test again.")
 def test_consume_error_store_offsets(kafka_cluster):
     """
     Tests to ensure that we handle messages with errors when storing offsets.
@@ -89,7 +106,7 @@ def test_consume_error_store_offsets(kafka_cluster):
     try:
         # Since the session timeout value is low, JoinGroupRequest will fail
         # and we get error in a message while polling.
-        m = consumer.poll(1)
+        m = consumer.poll(2)
         consumer.store_offsets(m)
     except KafkaException as e:
         assert e.args[0].code() == KafkaError._INVALID_ARG, \

tests/integration/consumer/test_consumer_memberid.py

@@ -16,6 +16,7 @@
 # limit
 
 import pytest
+from tests.common import TestUtils
 
 
 def test_consumer_memberid(kafka_cluster):
@@ -32,17 +33,29 @@ def test_consumer_memberid(kafka_cluster):
     consumer = kafka_cluster.consumer(consumer_conf)
 
     assert consumer is not None
-    assert consumer.memberid() is None
+    before_memberid = consumer.memberid()
+
+    # With implementation of KIP-1082, member id is generated on the client
+    # side for ConsumerGroupHeartbeat used in the `consumer` protocol
+    # introduced in KIP-848
+    if TestUtils.use_group_protocol_consumer():
+        assert before_memberid is not None
+        assert isinstance(before_memberid, str)
+        assert len(before_memberid) > 0
+    else:
+        assert before_memberid is None
+
     kafka_cluster.seed_topic(topic, value_source=[b'memberid'])
 
     consumer.subscribe([topic])
     msg = consumer.poll(10)
     assert msg is not None
     assert msg.value() == b'memberid'
-    memberid = consumer.memberid()
-    print("Member Id is -----> " + memberid)
-    assert isinstance(memberid, str)
-    assert len(memberid) > 0
+    after_memberid = consumer.memberid()
+    assert isinstance(after_memberid, str)
+    assert len(after_memberid) > 0
+    if TestUtils.use_group_protocol_consumer():
+        assert before_memberid == after_memberid
     consumer.close()
 
     with pytest.raises(RuntimeError) as error_info:

tests/integration/integration_test.py

@@ -333,9 +333,9 @@ def verify_producer_performance(with_dr_cb=True):
         bar.finish()
 
     print('# producing %d messages (%.2fMb) took %.3fs: %d msgs/s, %.2f Mb/s' %
-          (msgs_produced, bytecnt / (1024*1024), t_produce_spent,
+          (msgs_produced, bytecnt / (1024 * 1024), t_produce_spent,
            msgs_produced / t_produce_spent,
-           (bytecnt/t_produce_spent) / (1024*1024)))
+           (bytecnt / t_produce_spent) / (1024 * 1024)))
     print('# %d temporary produce() failures due to backpressure (local queue full)' % msgs_backpressure)
 
     print('waiting for %d/%d deliveries' % (len(p), msgs_produced))
@@ -344,9 +344,9 @@ def verify_producer_performance(with_dr_cb=True):
     t_delivery_spent = time.time() - t_produce_start
 
     print('# producing %d messages (%.2fMb) took %.3fs: %d msgs/s, %.2f Mb/s' %
-          (msgs_produced, bytecnt / (1024*1024), t_produce_spent,
+          (msgs_produced, bytecnt / (1024 * 1024), t_produce_spent,
            msgs_produced / t_produce_spent,
-           (bytecnt/t_produce_spent) / (1024*1024)))
+           (bytecnt / t_produce_spent) / (1024 * 1024)))
 
     # Fake numbers if not using a dr_cb
     if not with_dr_cb:
@@ -355,9 +355,9 @@ def verify_producer_performance(with_dr_cb=True):
         dr.bytes_delivered = bytecnt
 
     print('# delivering %d messages (%.2fMb) took %.3fs: %d msgs/s, %.2f Mb/s' %
-          (dr.msgs_delivered, dr.bytes_delivered / (1024*1024), t_delivery_spent,
+          (dr.msgs_delivered, dr.bytes_delivered / (1024 * 1024), t_delivery_spent,
            dr.msgs_delivered / t_delivery_spent,
-           (dr.bytes_delivered/t_delivery_spent) / (1024*1024)))
+           (dr.bytes_delivered / t_delivery_spent) / (1024 * 1024)))
     print('# post-produce delivery wait took %.3fs' %
           (t_delivery_spent - t_produce_spent))
 
@@ -447,7 +447,7 @@ def print_wmark(consumer, topic_parts):
         elif (msg.offset() % 4) == 0:
             offsets = c.commit(msg, asynchronous=False)
             assert len(offsets) == 1, 'expected 1 offset, not %s' % (offsets)
-            assert offsets[0].offset == msg.offset()+1, \
+            assert offsets[0].offset == msg.offset() + 1, \
                 'expected offset %d to be committed, not %s' % \
                 (msg.offset(), offsets)
             print('Sync committed offset: %s' % offsets)
@@ -515,7 +515,7 @@ def my_on_revoke(consumer, partitions):
         print('on_revoke:', len(partitions), 'partitions:')
         for p in partitions:
             print(' %s [%d] @ %d' % (p.topic, p.partition, p.offset))
-        consumer.unassign()
+        consumer.unassign(partitions)
 
     c.subscribe([topic], on_assign=my_on_assign, on_revoke=my_on_revoke)
 
@@ -559,8 +559,8 @@ def my_on_revoke(consumer, partitions):
     if msgcnt > 0:
         t_spent = time.time() - t_first_msg
         print('%d messages (%.2fMb) consumed in %.3fs: %d msgs/s, %.2f Mb/s' %
-              (msgcnt, bytecnt / (1024*1024), t_spent, msgcnt / t_spent,
-               (bytecnt / t_spent) / (1024*1024)))
+              (msgcnt, bytecnt / (1024 * 1024), t_spent, msgcnt / t_spent,
+               (bytecnt / t_spent) / (1024 * 1024)))
 
     print('closing consumer')
     c.close()
@@ -590,11 +590,11 @@ def verify_consumer_seek(c, seek_to_msg):
         print('seek: message at offset %d (epoch %d)' %
               (msg.offset(), msg.leader_epoch()))
         assert msg.offset() == seek_to_msg.offset() and \
-               msg.leader_epoch() == seek_to_msg.leader_epoch(), \
-               ('expected message at offset %d (epoch %d), ' % (seek_to_msg.offset(),
-                                                                seek_to_msg.leader_epoch())) + \
-               ('not %d (epoch %d)' % (msg.offset(),
-                                       msg.leader_epoch()))
+            msg.leader_epoch() == seek_to_msg.leader_epoch(), \
+            ('expected message at offset %d (epoch %d), ' % (seek_to_msg.offset(),
+                                                             seek_to_msg.leader_epoch())) + \
+            ('not %d (epoch %d)' % (msg.offset(),
+                                    msg.leader_epoch()))
         break
 
 
@@ -643,7 +643,7 @@ def verify_batch_consumer():
             elif (msg.offset() % 4) == 0:
                 offsets = c.commit(msg, asynchronous=False)
                 assert len(offsets) == 1, 'expected 1 offset, not %s' % (offsets)
-                assert offsets[0].offset == msg.offset()+1, \
+                assert offsets[0].offset == msg.offset() + 1, \
                     'expected offset %d to be committed, not %s' % \
                     (msg.offset(), offsets)
                 print('Sync committed offset: %s' % offsets)
@@ -697,7 +697,7 @@ def my_on_revoke(consumer, partitions):
         print('on_revoke:', len(partitions), 'partitions:')
         for p in partitions:
             print(' %s [%d] @ %d' % (p.topic, p.partition, p.offset))
-        consumer.unassign()
+        consumer.unassign(partitions)
 
     c.subscribe([topic], on_assign=my_on_assign, on_revoke=my_on_revoke)
 
@@ -738,8 +738,8 @@ def my_on_revoke(consumer, partitions):
     if msgcnt > 0:
         t_spent = time.time() - t_first_msg
         print('%d messages (%.2fMb) consumed in %.3fs: %d msgs/s, %.2f Mb/s' %
-              (msgcnt, bytecnt / (1024*1024), t_spent, msgcnt / t_spent,
-               (bytecnt / t_spent) / (1024*1024)))
+              (msgcnt, bytecnt / (1024 * 1024), t_spent, msgcnt / t_spent,
+               (bytecnt / t_spent) / (1024 * 1024)))
 
     print('closing consumer')
     c.close()
@@ -1035,8 +1035,8 @@ def stats_cb(stats_json_str):
     if msgcnt > 0:
         t_spent = time.time() - t_first_msg
         print('%d messages (%.2fMb) consumed in %.3fs: %d msgs/s, %.2f Mb/s' %
-              (msgcnt, bytecnt / (1024*1024), t_spent, msgcnt / t_spent,
-               (bytecnt / t_spent) / (1024*1024)))
+              (msgcnt, bytecnt / (1024 * 1024), t_spent, msgcnt / t_spent,
+               (bytecnt / t_spent) / (1024 * 1024)))
 
     print('closing consumer')
     c.close()