MINOR: Address flaky wait for producer acks logging for checking when not al… (#20957)

Adding the following for some system tests
- Increased timeout for `await_produced_messages`
- Added more logging for `test_fencing_consumer` which has exhibited
flakiness and is hard to reproduce

Reviewers: Lianet Magrans<[email protected]>

Author: bbejeck

tests/kafkatest/services/kafka/kafka.py

@@ -2023,3 +2023,29 @@ def get_offset_shell(self, time=None, topic=None, partitions=None, topic_partiti
 
     def java_class_name(self):
         return "kafka\.Kafka"
+
+    def describe_consumer_group_members(self, group, node=None, command_config=None):
+        """ Describe a consumer group.
+        """
+        if node is None:
+            node = self.nodes[0]
+        consumer_group_script = self.path.script("kafka-consumer-groups.sh", node)
+
+        if command_config is None:
+            command_config = ""
+        else:
+            command_config = "--command-config " + command_config
+
+        cmd = fix_opts_for_new_jvm(node)
+        cmd += "%s --bootstrap-server %s %s --group %s --describe" % \
+               (consumer_group_script,
+                self.bootstrap_servers(self.security_protocol),
+                command_config, group)
+
+        cmd += " --members"
+
+        output_lines = []
+        for line in node.account.ssh_capture(cmd):
+            if not (line.startswith("SLF4J") or line.startswith("GROUP") or line.strip() == ""):
+                output_lines.append(line.strip())
+        return output_lines

tests/kafkatest/tests/client/consumer_test.py

@@ -297,7 +297,7 @@ def test_fencing_static_consumer(self, num_conflict_consumers, fencing_stage, me
         producer = self.setup_producer(self.TOPIC)
 
         producer.start()
-        self.await_produced_messages(producer)
+        self.await_produced_messages(producer, timeout_sec=120)
 
         consumer = self.setup_consumer(self.TOPIC, static_membership=True, group_protocol=group_protocol)
 
@@ -324,18 +324,33 @@ def test_fencing_static_consumer(self, num_conflict_consumers, fencing_stage, me
                 # Consumer protocol: Existing members should remain active and new conflicting ones should not be able to join.
                 self.await_consumed_messages(consumer)
                 assert num_rebalances == consumer.num_rebalances(), "Static consumers attempt to join with instance id in use should not cause a rebalance"
-                assert len(consumer.joined_nodes()) == len(consumer.nodes)
+                try:
+                    assert len(consumer.joined_nodes()) == len(consumer.nodes)
+                except AssertionError:
+                    self.logger.debug("All members not in group %s. Describe output is %s", self.group_id,
+                                      " ".join(self.kafka.describe_consumer_group_members(self.group_id)))
+                    raise
                 assert len(conflict_consumer.joined_nodes()) == 0
-                
+
                 # Stop existing nodes, so conflicting ones should be able to join.
                 consumer.stop_all()
                 wait_until(lambda: len(consumer.dead_nodes()) == len(consumer.nodes),
                            timeout_sec=60,
-                           err_msg="Timed out waiting for the consumer to shutdown")
+                           err_msg="Timed out waiting for the consumer to shutdown. Describe output is %s" % " ".join(self.kafka.describe_consumer_group_members(self.group_id)))
+
+                # Wait until the group becomes empty to ensure the instance ID is released.
+                # We use the 60-second timeout because the consumer session timeout is 45 seconds adding some time for latency.
+                wait_until(lambda: self.group_id in self.kafka.list_consumer_groups(state="empty"),
+                           timeout_sec=60,
+                           err_msg="Timed out waiting for the consumers to be removed from the group. Describe output is %s." % " ".join(self.kafka.describe_consumer_group_members(self.group_id)))
+
                 conflict_consumer.start()
-                self.await_members(conflict_consumer, num_conflict_consumers)
+                try:
+                    self.await_members(conflict_consumer, num_conflict_consumers)
+                except TimeoutError:
+                    self.logger.debug("All conflict members not in group %s. Describe output is %s", self.group_id, " ".join(self.kafka.describe_consumer_group_members(self.group_id)))
+                    raise
 
-            
         else:
             consumer.start()
             conflict_consumer.start()