Repository: apache/kafka
File: tests/kafkatest/tests/streams/streams_broker_down_resilience_test.py (Python)

Code before:

        with processor.node.account.monitor_log(processor.LOG_FILE) as monitor_1:
            with processor_2.node.account.monitor_log(processor_2.LOG_FILE) as monitor_2:
                with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor_3:
                    self.kafka.start_node(node)

                    monitor_1.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor.node.account))
                    monitor_2.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_2.node.account))
                    monitor_3.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_3.node.account))

        with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:
            with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
                with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

                    self.assert_produce(self.inputTopic,
                                        "sending_message_after_broker_down_initially",
                                        num_messages=self.num_messages,
                                        timeout_sec=120)

                    monitor_1.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
                    monitor_2.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
                    monitor_3.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

                    self.assert_consume(self.client_id,
                                        "consuming_message_after_broker_down_initially",
                                        self.outputTopic,
                                        num_messages=self.num_messages,
                                        timeout_sec=120)

        self.kafka.stop()

    @cluster(num_nodes=9)
    @matrix(metadata_quorum=[quorum.combined_kraft],
            group_protocol=["classic", "streams"])
    def test_streams_should_scale_in_while_brokers_down(self, metadata_quorum, group_protocol):
        self.kafka.start()
        extra_configs = ",application.id=shutdown_with_broker_down"

        # TODO KIP-441: consider rewriting the test for HighAvailabilityTaskAssignor

Review comment (refactor):
[nitpick] This comment is repeated verbatim across multiple test cases. Consider abstracting the key-production logic into a helper or adding a shared comment to reduce duplication.

Code after:

        with processor.node.account.monitor_log(processor.LOG_FILE) as monitor_1:
            with processor_2.node.account.monitor_log(processor_2.LOG_FILE) as monitor_2:
                with processor_3.node.account.monitor_log(processor_3.LOG_FILE) as monitor_3:
                    self.kafka.start_node(node)

                    monitor_1.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor.node.account))
                    monitor_2.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_2.node.account))
                    monitor_3.wait_until(self.connected_message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.connected_message) + str(processor_3.node.account))

        with processor.node.account.monitor_log(processor.STDOUT_FILE) as monitor_1:
            with processor_2.node.account.monitor_log(processor_2.STDOUT_FILE) as monitor_2:
                with processor_3.node.account.monitor_log(processor_3.STDOUT_FILE) as monitor_3:

                    # repeating_keys enables production of records with keys, ensuring that we produce to all 3 partitions
                    self.assert_produce(self.inputTopic,
                                        "sending_message_after_broker_down_initially",
                                        num_messages=self.num_messages,
                                        timeout_sec=120,
                                        repeating_keys=self.num_messages)

                    monitor_1.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor.node.account))
                    monitor_2.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor_2.node.account))
                    monitor_3.wait_until(self.message,
                                         timeout_sec=120,
                                         err_msg=("Never saw '%s' on " % self.message) + str(processor_3.node.account))

                    self.assert_consume(self.client_id,
                                        "consuming_message_after_broker_down_initially",
                                        self.outputTopic,
                                        num_messages=self.num_messages,
                                        timeout_sec=120)

        self.kafka.stop()

    @cluster(num_nodes=9)
    @matrix(metadata_quorum=[quorum.combined_kraft],
            group_protocol=["classic", "streams"])
    def test_streams_should_scale_in_while_brokers_down(self, metadata_quorum, group_protocol):
        self.kafka.start()
        extra_configs = ",application.id=shutdown_with_broker_down"