Repository: apache/kafka
File: tests/kafkatest/tests/client/consumer_test.py (Python)

Code before:
                                      set(str(node.account) for node in consumer.nodes) - set(consumer.joined_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

                # Conflict consumers will terminate due to a fatal UnreleasedInstanceIdException error.
                # Wait for termination to complete to prevent conflict consumers from immediately re-joining the group while existing nodes are shutting down.
                self.await_conflict_consumers_fenced(conflict_consumer)

                # 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. 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()
                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()

            wait_until(lambda: len(consumer.joined_nodes()) + len(conflict_consumer.joined_nodes()) == len(consumer.nodes),
                       timeout_sec=60,
                       err_msg="Timed out waiting for consumers to join, expected total %d joined, but only see %d joined from "
                               "normal consumer group and %d from conflict consumer group" % \
                               (len(consumer.nodes), len(consumer.joined_nodes()), len(conflict_consumer.joined_nodes()))
                       )
            wait_until(lambda: len(consumer.dead_nodes()) + len(conflict_consumer.dead_nodes()) == len(conflict_consumer.nodes),
                       timeout_sec=60,
                       err_msg="Timed out waiting for fenced consumers to die, expected total %d dead, but only see %d dead in "
                               "normal consumer group and %d dead in conflict consumer group" % \
                               (len(conflict_consumer.nodes), len(consumer.dead_nodes()), len(conflict_consumer.dead_nodes()))
                       )

Review comment (suggestion):
```suggestion
                           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)))
```

Code after:
                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

                # Conflict consumers will terminate due to a fatal UnreleasedInstanceIdException error.
                # Wait for termination to complete to prevent conflict consumers from immediately re-joining the group while existing nodes are shutting down.
                self.await_conflict_consumers_fenced(conflict_consumer)

                # 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. 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()
                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()

            wait_until(lambda: len(consumer.joined_nodes()) + len(conflict_consumer.joined_nodes()) == len(consumer.nodes),
                       timeout_sec=60,
                       err_msg="Timed out waiting for consumers to join, expected total %d joined, but only see %d joined from "
                               "normal consumer group and %d from conflict consumer group" % \
                               (len(consumer.nodes), len(consumer.joined_nodes()), len(conflict_consumer.joined_nodes()))
                       )
            wait_until(lambda: len(consumer.dead_nodes()) + len(conflict_consumer.dead_nodes()) == len(conflict_consumer.nodes),
                       timeout_sec=60,
                       err_msg="Timed out waiting for fenced consumers to die, expected total %d dead, but only see %d dead in "
                               "normal consumer group and %d dead in conflict consumer group" % \
                               (len(conflict_consumer.nodes), len(consumer.dead_nodes()), len(conflict_consumer.dead_nodes()))
                       )

    @cluster(num_nodes=7)
    @matrix(
        clean_shutdown=[True],
        enable_autocommit=[True, False],
        metadata_quorum=[quorum.isolated_kraft],