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

Code before:
        # The consumers' session timeouts must exceed the time it takes for a broker to roll.  Consumers are likely
        # to see cluster metadata consisting of just a single alive broker in the case where the cluster has just 2
        # brokers and the cluster is rolling (which is what is happening here).  When the consumer sees a single alive
        # broker, and then that broker rolls, the consumer will be unable to connect to the cluster until that broker
        # completes its roll.  In the meantime, the consumer group will move to the group coordinator on the other
        # broker, and that coordinator will fail the consumer and trigger a group rebalance if its session times out.
        # This test is asserting that no rebalances occur, so we increase the session timeout for this to be the case.
        self.session_timeout_sec = 30
        consumer = self.setup_consumer(self.TOPIC, group_protocol=group_protocol)

        producer.start()
        self.await_produced_messages(producer)

        consumer.start()
        self.await_all_members(consumer)

        num_rebalances = consumer.num_rebalances()
        # TODO: make this test work with hard shutdowns, which probably requires
        #       pausing before the node is restarted to ensure that any ephemeral
        #       nodes have time to expire
        self.rolling_bounce_brokers(consumer, clean_shutdown=True)

        if group_protocol == consumer_group.classic_group_protocol:
            unexpected_rebalances = consumer.num_rebalances() - num_rebalances
            assert unexpected_rebalances == 0, \
                "Broker rolling bounce caused %d unexpected group rebalances" % unexpected_rebalances

        consumer.stop_all()

        assert consumer.current_position(partition) == consumer.total_consumed(), \
            "Total consumed records %d did not match consumed position %d" % \
            (consumer.total_consumed(), consumer.current_position(partition))

    @cluster(num_nodes=7)
    @matrix(
        clean_shutdown=[True],
        bounce_mode=["all", "rolling"],
        metadata_quorum=[quorum.zk, quorum.isolated_kraft],
        use_new_coordinator=[False]
    )
    @matrix(
        clean_shutdown=[True],
        bounce_mode=["all", "rolling"],
        metadata_quorum=[quorum.isolated_kraft],
        use_new_coordinator=[True],
        group_protocol=consumer_group.all_group_protocols
    )
    def test_consumer_bounce(self, clean_shutdown, bounce_mode, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
        """
        Verify correct consumer behavior when the consumers in the group are consecutively restarted.


Review comment (suggestion):
The [current framework](https://github.com/apache/kafka/blob/0a4a12fbc4df0ac4b141a869d3880b15b66a7759/tests/kafkatest/services/verifiable_consumer.py#L454-L456) calculates `num_rebalances` by increment by one every time a new assignment is received.

For new protocol consumer, if reconciliation happens during bouncing the brokers, `num_rebalances` will also be incremented. Currently there's no way to figure out whether a new rebalance is triggered or a reconciliation happens, so the assertion is skipped here.

Code after:
        # The consumers' session timeouts must exceed the time it takes for a broker to roll.  Consumers are likely
        # to see cluster metadata consisting of just a single alive broker in the case where the cluster has just 2
        # brokers and the cluster is rolling (which is what is happening here).  When the consumer sees a single alive
        # broker, and then that broker rolls, the consumer will be unable to connect to the cluster until that broker
        # completes its roll.  In the meantime, the consumer group will move to the group coordinator on the other
        # broker, and that coordinator will fail the consumer and trigger a group rebalance if its session times out.
        # This test is asserting that no rebalances occur, so we increase the session timeout for this to be the case.
        self.session_timeout_sec = 30
        consumer = self.setup_consumer(self.TOPIC, group_protocol=group_protocol)

        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)

        num_rebalances = consumer.num_rebalances()
        # TODO: make this test work with hard shutdowns, which probably requires
        #       pausing before the node is restarted to ensure that any ephemeral
        #       nodes have time to expire
        self.rolling_bounce_brokers(consumer, clean_shutdown=True)

        unexpected_rebalances = consumer.num_rebalances() - num_rebalances
        assert unexpected_rebalances == 0, \
            "Broker rolling bounce caused %d unexpected group rebalances" % unexpected_rebalances

        consumer.stop_all()

        assert consumer.current_position(partition) == consumer.total_consumed(), \
            "Total consumed records %d did not match consumed position %d" % \
            (consumer.total_consumed(), consumer.current_position(partition))

    @cluster(num_nodes=7)
    @matrix(
        clean_shutdown=[True],
        bounce_mode=["all", "rolling"],
        metadata_quorum=[quorum.zk, quorum.isolated_kraft],
        use_new_coordinator=[False]
    )
    @matrix(
        clean_shutdown=[True],
        bounce_mode=["all", "rolling"],
        metadata_quorum=[quorum.isolated_kraft],
        use_new_coordinator=[True],
        group_protocol=consumer_group.all_group_protocols
    )
    def test_consumer_bounce(self, clean_shutdown, bounce_mode, metadata_quorum=quorum.zk, use_new_coordinator=False, group_protocol=None):
        """
        Verify correct consumer behavior when the consumers in the group are consecutively restarted.