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

Code before:
                self.await_all_members(consumer)
                self.await_consumed_messages(consumer)

    def setup_consumer(self, topic, **kwargs):
        # collect verifiable consumer events since this makes debugging much easier
        consumer = super(OffsetValidationTest, self).setup_consumer(topic, **kwargs)
        self.mark_for_collect(consumer, 'verifiable_consumer_stdout')
        return consumer

    def await_conflict_consumers_fenced(self, conflict_consumer):
        # Ensure every conflicting consumer actually starts once before we wait for fencing.
        started_nodes = set()
        def all_conflict_consumers_started():
            for node in conflict_consumer.alive_nodes():
                started_nodes.add(node)
            return len(conflict_consumer.alive_nodes()) == len(conflict_consumer.nodes)

        wait_until(all_conflict_consumers_started,
                   timeout_sec=60,
                   err_msg="Timed out waiting for conflict consumers to start before fencing")

        wait_until(lambda: len(conflict_consumer.dead_nodes()) == len(conflict_consumer.nodes),
                   timeout_sec=60,
                   err_msg="Timed out waiting for conflict consumers to terminate after fencing")

        # Guard against stray processes that could rejoin once the original static members stop.
        running_pids = {}
        def all_conflict_consumers_stopped():
            running_pids.clear()
            for node in conflict_consumer.nodes:
                pids = conflict_consumer.pids(node)
                if pids:
                    running_pids[node.account.hostname] = pids
            return not running_pids

        wait_until(all_conflict_consumers_stopped,
                   timeout_sec=30,
                   err_msg=lambda: "Conflict consumers still running after fencing: %s" %
                                   ", ".join("%s=%s" % (host, pids) for host, pids in running_pids.items()))

    @cluster(num_nodes=7)
    @matrix(
        metadata_quorum=[quorum.isolated_kraft],
        group_protocol=consumer_group.all_group_protocols
    )
    def test_broker_rolling_bounce(self, metadata_quorum=quorum.isolated_kraft, group_protocol=None):
        """
        Verify correct consumer behavior when the brokers are consecutively restarted.

        Setup: single Kafka cluster with one producer writing messages to a single topic with one
        partition, a set of consumers in the same group reading from the same topic.

Review comment (bug):
Could you remind readers that `UnreleasedInstanceIdException` is a fatal error, resulting in the consumer's immediate failure

Code after:
                self.await_all_members(consumer)
                self.await_consumed_messages(consumer)

    def setup_consumer(self, topic, **kwargs):
        # collect verifiable consumer events since this makes debugging much easier
        consumer = super(OffsetValidationTest, self).setup_consumer(topic, **kwargs)
        self.mark_for_collect(consumer, 'verifiable_consumer_stdout')
        return consumer

    def await_conflict_consumers_fenced(self, conflict_consumer):
        # Rely on explicit shutdown_complete events from the verifiable consumer to guarantee each conflict member
        # reached the fenced path rather than remaining in the default DEAD state prior to startup.
        wait_until(lambda: len(conflict_consumer.shutdown_complete_nodes()) == len(conflict_consumer.nodes) and 
                           len(conflict_consumer.dead_nodes()) == len(conflict_consumer.nodes),
                   timeout_sec=60,
                   err_msg="Timed out waiting for conflict consumers to report shutdown completion after fencing")
        
    @cluster(num_nodes=7)
    @matrix(
        metadata_quorum=[quorum.isolated_kraft],
        group_protocol=consumer_group.all_group_protocols
    )
    def test_broker_rolling_bounce(self, metadata_quorum=quorum.isolated_kraft, group_protocol=None):
        """
        Verify correct consumer behavior when the brokers are consecutively restarted.

        Setup: single Kafka cluster with one producer writing messages to a single topic with one
        partition, a set of consumers in the same group reading from the same topic.

        - Start a producer which continues producing new messages throughout the test.
        - Start up the consumers and wait until they've joined the group.
        - In a loop, restart each broker consecutively, waiting for the group to stabilize between
          each broker restart.
        - Verify delivery semantics according to the failure type and that the broker bounces
          did not cause unexpected group rebalances.
        """
        partition = TopicPartition(self.TOPIC, 0)

        producer = self.setup_producer(self.TOPIC)
        # Due to KIP-899, which rebootstrap is performed when there are no available brokers in the current metadata.
        # We disable rebootstrapping by setting `metadata.recovery.strategy=none` for the consumer, as the test expects no metadata changes.
        # see KAFKA-18194
        consumer = self.setup_consumer(self.TOPIC, group_protocol=group_protocol, prop_file="metadata.recovery.strategy=none")

        producer.start()
        self.await_produced_messages(producer)

        consumer.start()
        self.await_all_members(consumer)
        self.await_all_members_stabilized(self.TOPIC, self.NUM_PARTITIONS, consumer, timeout_sec=60)