Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java (Java)

Code before:
        kafkaStreams.start();

        final long currentTimeNew = CLUSTER.time.milliseconds();

        processKeyValueAndVerify(
            key,
            value,
            currentTimeNew,
            List.of(
                new KeyValueTimestamp<>(key, value, currentTimeNew)
            )
        );

        kafkaStreams.close();
        kafkaStreams = null;
    }


    private <K, V> void processKeyValueAndVerify(
        final K key,
        final V value,
        final long timestamp,
        final List<KeyValueTimestamp<K, V>> expected)
        throws Exception {

        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            singletonList(KeyValue.pair(key, value)),
            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
                                     StringSerializer.class,
                                     StringSerializer.class),
            timestamp);


        final Properties consumerProperties = new Properties();
        consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group-" + safeTestName);
        consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());


        final List<KeyValueTimestamp<K, V>> actual =
            IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
            consumerProperties,
            outputTopic,
            expected.size(),
            60 * 1000);

        assertThat(actual, is(expected));


Review comment (suggestion):
The KafkaStreams instance should be closed with a timeout to prevent hanging tests. Consider using kafkaStreams.close(Duration.ofSeconds(30)) as done in the shutdown() method.
```suggestion
        kafkaStreams.close(Duration.ofSeconds(30));
```

Code after:

        props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
        processExactlyOneRecord(streamsBuilder, props, "3", "C");
    }

    private void processExactlyOneRecord(
        final StreamsBuilder streamsBuilder,
        final Properties props,
        final String key,
        final String value)
        throws Exception {
        kafkaStreams = new KafkaStreams(streamsBuilder.build(), props);
        kafkaStreams.start();

        final long currentTimeNew = CLUSTER.time.milliseconds();

        processKeyValueAndVerify(
            key,
            value,
            currentTimeNew,
            List.of(
                new KeyValueTimestamp<>(key, value, currentTimeNew)
            )
        );

        kafkaStreams.close();
        kafkaStreams = null;
    }


    private <K, V> void processKeyValueAndVerify(
        final K key,
        final V value,
        final long timestamp,
        final List<KeyValueTimestamp<K, V>> expected)
        throws Exception {

        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            singletonList(KeyValue.pair(key, value)),
            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
                StringSerializer.class,
                StringSerializer.class),
            timestamp);


        final Properties consumerProperties = new Properties();
        consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group-" + safeTestName);
        consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());