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

Code before:

        // wipe out state store to trigger restore process on restart
        streams.close();
        streams.cleanUp();

        // Restart the stream instance. There should not be exception handling the null
        // value within changelog topic.
        final List<KeyValue<Integer, Bytes>> newKeyValues = Collections
                .singletonList(KeyValue.pair(2, new Bytes(new byte[3])));
        IntegrationTestUtils.produceKeyValuesSynchronously(
                inputTopic, newKeyValues, producerConfig, new MockTime());
        streams = new KafkaStreams(builder.build(streamsConfiguration), streamsConfiguration);
        streams.start();
        IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
                consumerConfig, outputTopic, newKeyValues);
        streams.close();
    }

    @ParameterizedTest
    @CsvSource({
        "true,true",
        "true,false", 
        "false,true",
        "false,false"
    })
    public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
        final AtomicInteger numReceived = new AtomicInteger(0);
        final Topology topology = new Topology();

        final Properties props = props(stateUpdaterEnabled);

        // restoring from 1000 to 4000 (committed), and then process from 4000 to 5000 on each of the two partitions
        final int offsetLimitDelta = 1000;
        final int offsetCheckpointed = 1000;
        createStateForRestoration(inputStream, 0);
        setCommittedOffset(inputStream, offsetLimitDelta, useNewProtocol);

        final StateDirectory stateDirectory = new StateDirectory(new StreamsConfig(props), new MockTime(), true, false);
        // note here the checkpointed offset is the last processed record's offset, so without control message we should write this offset - 1
        new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 0)), ".checkpoint"))
            .write(Collections.singletonMap(new TopicPartition(inputStream, 0), (long) offsetCheckpointed - 1));
        new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 1)), ".checkpoint"))
            .write(Collections.singletonMap(new TopicPartition(inputStream, 1), (long) offsetCheckpointed - 1));

        final CountDownLatch startupLatch = new CountDownLatch(1);
        final CountDownLatch shutdownLatch = new CountDownLatch(1);

        topology.addReadOnlyStateStore(
            Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore("store"),
                new Serdes.IntegerSerde(),

Review comment (suggestion):
We should do the same for all other test case, like `shouldRestoreNullRecord` above. There is 8 test methods I did not miscount, but you enable the new protocol only for two.

Code after:

        final Properties consumerConfig = TestUtils.consumerConfig(
                CLUSTER.bootstrapServers(), IntegerDeserializer.class, BytesDeserializer.class);

        IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
                consumerConfig, outputTopic, initialKeyValues);

        // wipe out state store to trigger restore process on restart
        streams.close();
        streams.cleanUp();

        // Restart the stream instance. There should not be exception handling the null
        // value within changelog topic.
        final List<KeyValue<Integer, Bytes>> newKeyValues = Collections
                .singletonList(KeyValue.pair(2, new Bytes(new byte[3])));
        IntegrationTestUtils.produceKeyValuesSynchronously(
                inputTopic, newKeyValues, producerConfig, new MockTime());
        streams = new KafkaStreams(builder.build(streamsConfiguration), streamsConfiguration);
        streams.start();
        IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
                consumerConfig, outputTopic, newKeyValues);
        streams.close();
    }

    @ParameterizedTest
    @CsvSource({
        "true,true",
        "true,false", 
        "false,true",
        "false,false"
    })
    public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
        final AtomicInteger numReceived = new AtomicInteger(0);
        final Topology topology = new Topology();

        final Properties props = props(stateUpdaterEnabled);
        if (useNewProtocol) {
            props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
        }

        // restoring from 1000 to 4000 (committed), and then process from 4000 to 5000 on each of the two partitions
        final int offsetLimitDelta = 1000;
        final int offsetCheckpointed = 1000;
        createStateForRestoration(inputStream, 0);
        if (!useNewProtocol) {
            setCommittedOffset(inputStream, offsetLimitDelta, useNewProtocol);
        }

        final StateDirectory stateDirectory = new StateDirectory(new StreamsConfig(props), new MockTime(), true, false);
        // note here the checkpointed offset is the last processed record's offset, so without control message we should write this offset - 1
        new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(new TaskId(0, 0)), ".checkpoint"))