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

Code before:
        properties = mkObjectProperties(
            mkMap(
                mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),
                mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()),
                mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2),
                mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class),
                mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class),
                mkEntry(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000)
            )
        );
    }

    private void produceTestRecords(final String inputTopic, final EmbeddedKafkaCluster cluster) {
        final Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "test-client");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        try (KafkaProducer<String, String> dummyProducer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 5000; i++) {
                final String key = "key-" + (i % 50);
                final String value = "value-" + i;
                dummyProducer.send(new ProducerRecord<>(inputTopic, key, value));
            }
            dummyProducer.close();
        } 
    }

    private void startStreamsAndWaitForRunning(final KafkaStreams kafkaStreams) throws InterruptedException {
        kafkaStreams.start();
        waitForRunning();
    }

    @AfterEach
    public void teardown() throws IOException {
        purgeLocalStreamsState(properties);
    }

    private void addStreamStateChangeListener(final KafkaStreams kafkaStreams) {
        kafkaStreams.setStateListener(
            (newState, oldState) -> stateTransitionHistory.add(newState)
        );
    }

    private void waitForRunning() throws InterruptedException {
        waitForCondition(
            () -> !stateTransitionHistory.isEmpty() &&
                stateTransitionHistory.get(stateTransitionHistory.size() - 1).equals(KafkaStreams.State.RUNNING),
            DEFAULT_DURATION.toMillis(),
            () -> String.format("Client did not transit to state %s in %d seconds",

Review comment (suggestion):
try-with-resouce does close the producer automatically . That's why we want to use it :) this line can be removed

Code after:
        properties = mkObjectProperties(
            mkMap(
                mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),
                mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()),
                mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2),
                mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class),
                mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class),
                mkEntry(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000)
            )
        );
    }

    private void produceTestRecords(final String inputTopic, final EmbeddedKafkaCluster cluster) {
        final Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "test-client");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 1000; i++) {
                final String key = "key-" + (i % 50);
                final String value = "value-" + i;
                producer.send(new ProducerRecord<>(inputTopic, key, value));
            }
        } 
    }

    private void startStreamsAndWaitForRunning(final KafkaStreams kafkaStreams) throws InterruptedException {
        kafkaStreams.start();
        waitForRunning();
    }

    @AfterEach
    public void teardown() throws IOException {
        purgeLocalStreamsState(properties);
    }

    private void addStreamStateChangeListener(final KafkaStreams kafkaStreams) {
        kafkaStreams.setStateListener(
            (newState, oldState) -> stateTransitionHistory.add(newState)
        );
    }

    private void waitForRunning() throws InterruptedException {
        waitForCondition(
            () -> !stateTransitionHistory.isEmpty() &&
                stateTransitionHistory.get(stateTransitionHistory.size() - 1).equals(KafkaStreams.State.RUNNING),
            DEFAULT_DURATION.toMillis(),
            () -> String.format("Client did not transit to state %s in %d seconds",
                KafkaStreams.State.RUNNING, DEFAULT_DURATION.toMillis() / 1000)