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

Code before:

                self.logger.debug("Members status - Joined [%d/%d]: %s, Not joined: %s",
                                  len(consumer.joined_nodes()),
                                  len(consumer.nodes),
                                  " ".join(str(node.account) for node in consumer.joined_nodes()),
                                  " ".join(set(str(node.account) for node in consumer.nodes) - set(consumer.joined_nodes())))

                if len(consumer.joined_nodes()) != len(consumer.nodes):
                    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)))

                assert len(consumer.joined_nodes()) == len(consumer.nodes)
                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")
                # Wait until the group becomes empty to ensure the instance ID is released.
                self.logger.debug("Stopped consumers waiting for group %s to be removed from the group", self.group_id)

                if len(consumer.joined_nodes()) != 0:
                    self.logger.debug("Not all stopped consumers removed %s. Describe output is %s", self.group_id, " ".join(self.kafka.describe_consumer_group_members(self.group_id)))

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

                if len(consumer.joined_nodes()) != num_conflict_consumers:
                    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)))

                self.await_members(conflict_consumer, num_conflict_consumers)

            
        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),

Review comment (bug):
can this condition be true at this point? (I would imagine that if ever there were still `joined_nodes`, the condition above on `wait_until(lambda: len(consumer.dead_nodes()) == len(consumer.nodes)` would fail before hitting this line, but I could be missing something).

If what I imagine is the case, then I guess we should wait in a try/except to be able to log what we want in the catch maybe?

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

    @cluster(num_nodes=7)
    @matrix(
        clean_shutdown=[True],