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

Code before:
                Integer::sum,
                Materialized.<Integer, Integer, KeyValueStore<Bytes, byte[]>>as("reduce-store").withLoggingDisabled()
            );
        final Properties props = props(stateUpdaterEnabled);
        if (useNewProtocol) {
            props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
        }
        kafkaStreams = new KafkaStreams(builder.build(), props);
        try {
            startApplicationAndWaitUntilRunning(kafkaStreams);
        } catch (final Exception e) {
            fail("Failed to start KafkaStreams", e);
        }
    }

    @ParameterizedTest
    @CsvSource({
        "true,true",
        "true,false",
        "false,true",
        "false,false"
    })
    public void shouldProcessDataFromStoresWithLoggingDisabled(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws InterruptedException {
        IntegrationTestUtils.produceKeyValuesSynchronously(inputStream,
                asList(KeyValue.pair(1, 1),
                        KeyValue.pair(2, 2),
                        KeyValue.pair(3, 3)),
                TestUtils.producerConfig(CLUSTER.bootstrapServers(),
                        IntegerSerializer.class,
                        IntegerSerializer.class),
                CLUSTER.time);

        final KeyValueBytesStoreSupplier lruMapSupplier = Stores.lruMap(inputStream, 10);

        final StoreBuilder<KeyValueStore<Integer, Integer>> storeBuilder = new KeyValueStoreBuilder<>(lruMapSupplier,
                Serdes.Integer(),
                Serdes.Integer(),
                CLUSTER.time)
                .withLoggingDisabled();

        final StreamsBuilder streamsBuilder = new StreamsBuilder();

        streamsBuilder.addStateStore(storeBuilder);

        final KStream<Integer, Integer> stream = streamsBuilder.stream(inputStream);
        final CountDownLatch processorLatch = new CountDownLatch(3);
        stream.process(() -> new KeyValueStoreProcessor(inputStream, processorLatch), inputStream);

        final Topology topology = streamsBuilder.build();

        final Properties props = props(stateUpdaterEnabled);

Review comment (suggestion):
Nit: I noticed a few repeated snippets like
```java
if (useNewProtocol) {
        props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
    }
```

Maybe it could be helpful to add small overloads
```java
private Properties props(final boolean useNewProtocol) {
    return props(mkObjectProperties(mkMap()), useNewProtocol);
}

private Properties props(final Properties extraProperties, final boolean useNewProtocol) {
    final Properties streamsConfiguration = props(extraProperties);
    if (useNewProtocol) {
        streamsConfiguration.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
    }
    return streamsConfiguration;
}
```
Then call sites could just use `props(useNewProtocol)` or `props(extra, useNewProtocol)` to reduce the repeated if blocks.


Code after:
                TestUtils.producerConfig(CLUSTER.bootstrapServers(),
                        IntegerSerializer.class,
                        IntegerSerializer.class),
                CLUSTER.time);

        final KeyValueBytesStoreSupplier lruMapSupplier = Stores.lruMap(inputStream, 10);

        final StoreBuilder<KeyValueStore<Integer, Integer>> storeBuilder = new KeyValueStoreBuilder<>(lruMapSupplier,
                Serdes.Integer(),
                Serdes.Integer(),
                CLUSTER.time)
                .withLoggingDisabled();

        final StreamsBuilder streamsBuilder = new StreamsBuilder();

        streamsBuilder.addStateStore(storeBuilder);

        final KStream<Integer, Integer> stream = streamsBuilder.stream(inputStream);
        final CountDownLatch processorLatch = new CountDownLatch(3);
        stream.process(() -> new KeyValueStoreProcessor(inputStream, processorLatch), inputStream);

        final Topology topology = streamsBuilder.build();

        final Properties props = props();

        if (useNewProtocol) {
            props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
        }
        kafkaStreams = new KafkaStreams(topology, props);

        final CountDownLatch latch = new CountDownLatch(1);
        kafkaStreams.setStateListener((newState, oldState) -> {
            if (newState == KafkaStreams.State.RUNNING && oldState == KafkaStreams.State.REBALANCING) {
                latch.countDown();
            }
        });
        kafkaStreams.start();

        latch.await(30, TimeUnit.SECONDS);

        assertTrue(processorLatch.await(30, TimeUnit.SECONDS));
    }

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void shouldRecycleStateFromStandbyTaskPromotedToActiveTaskAndNotRestore(final boolean useNewProtocol) throws Exception {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table(
                inputStream,
                Consumed.with(Serdes.Integer(), Serdes.Integer()), Materialized.as(getCloseCountingStore("store"))
        );