KAFKA-19666: Remove old restoration codepath from RestoreIntegrationTest [4/N] (#20498)

Clean up `RestoreIntegrationTest.java`

Reviewers: Lucas Brutschy <[email protected]>

Author: shashankhs11

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java

@@ -43,7 +43,6 @@
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.StreamsConfig.InternalConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
 import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
@@ -77,7 +76,6 @@
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -161,8 +159,8 @@ public void createTopics(final TestInfo testInfo) throws InterruptedException {
         CLUSTER.createTopic(inputStream, 2, 1);
     }
 
-    private Properties props(final boolean stateUpdaterEnabled) {
-        return props(mkObjectProperties(mkMap(mkEntry(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled))));
+    private Properties props() {
+        return props(mkObjectProperties(mkMap()));
     }
 
     private Properties props(final Properties extraProperties) {
@@ -267,17 +265,12 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol) throws Excepti
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false", 
-        "false,true",
-        "false,false"
-    })
-    public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
+    @ValueSource(booleans = {true, false})
+    public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean useNewProtocol) throws Exception {
         final AtomicInteger numReceived = new AtomicInteger(0);
         final Topology topology = new Topology();
 
-        final Properties props = props(stateUpdaterEnabled);
+        final Properties props = props();
         if (useNewProtocol) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
         }
@@ -338,17 +331,12 @@ public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stat
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false",
-        "false,true",
-        "false,false"
-    })
-    public void shouldRestoreStateFromSourceTopicForGlobalTable(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
+    @ValueSource(booleans = {true, false})
+    public void shouldRestoreStateFromSourceTopicForGlobalTable(final boolean useNewProtocol) throws Exception {
         final AtomicInteger numReceived = new AtomicInteger(0);
         final StreamsBuilder builder = new StreamsBuilder();
 
-        final Properties props = props(stateUpdaterEnabled);
+        final Properties props = props();
         props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
         if (useNewProtocol) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
@@ -413,20 +401,15 @@ public void shouldRestoreStateFromSourceTopicForGlobalTable(final boolean stateU
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false",
-        "false,true",
-        "false,false"
-    })
-    public void shouldRestoreStateFromChangelogTopic(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
+    @ValueSource(booleans = {true, false})
+    public void shouldRestoreStateFromChangelogTopic(final boolean useNewProtocol) throws Exception {
         final String changelog = appId + "-store-changelog";
         CLUSTER.createTopic(changelog, 2, 1);
 
         final AtomicInteger numReceived = new AtomicInteger(0);
         final StreamsBuilder builder = new StreamsBuilder();
 
-        final Properties props = props(stateUpdaterEnabled);
+        final Properties props = props();
 
         if (useNewProtocol) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
@@ -474,13 +457,8 @@ public void shouldRestoreStateFromChangelogTopic(final boolean stateUpdaterEnabl
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false",
-        "false,true",
-        "false,false"
-    })
-    public void shouldSuccessfullyStartWhenLoggingDisabled(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws InterruptedException {
+    @ValueSource(booleans = {true, false})
+    public void shouldSuccessfullyStartWhenLoggingDisabled(final boolean useNewProtocol) throws InterruptedException {
         final StreamsBuilder builder = new StreamsBuilder();
 
         final KStream<Integer, Integer> stream = builder.stream(inputStream);
@@ -490,7 +468,7 @@ public void shouldSuccessfullyStartWhenLoggingDisabled(final boolean stateUpdate
                 Integer::sum,
                 Materialized.<Integer, Integer, KeyValueStore<Bytes, byte[]>>as("reduce-store").withLoggingDisabled()
             );
-        final Properties props = props(stateUpdaterEnabled);
+        final Properties props = props();
         if (useNewProtocol) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
         }
@@ -503,13 +481,8 @@ public void shouldSuccessfullyStartWhenLoggingDisabled(final boolean stateUpdate
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false",
-        "false,true",
-        "false,false"
-    })
-    public void shouldProcessDataFromStoresWithLoggingDisabled(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws InterruptedException {
+    @ValueSource(booleans = {true, false})
+    public void shouldProcessDataFromStoresWithLoggingDisabled(final boolean useNewProtocol) throws InterruptedException {
         IntegrationTestUtils.produceKeyValuesSynchronously(inputStream,
                 asList(KeyValue.pair(1, 1),
                         KeyValue.pair(2, 2),
@@ -537,7 +510,7 @@ public void shouldProcessDataFromStoresWithLoggingDisabled(final boolean stateUp
 
         final Topology topology = streamsBuilder.build();
 
-        final Properties props = props(stateUpdaterEnabled);
+        final Properties props = props();
 
         if (useNewProtocol) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
@@ -558,13 +531,8 @@ public void shouldProcessDataFromStoresWithLoggingDisabled(final boolean stateUp
     }
 
     @ParameterizedTest
-    @CsvSource({
-        "true,true",
-        "true,false",
-        "false,true",
-        "false,false"
-    })
-    public void shouldRecycleStateFromStandbyTaskPromotedToActiveTaskAndNotRestore(final boolean stateUpdaterEnabled, final boolean useNewProtocol) throws Exception {
+    @ValueSource(booleans = {true, false})
+    public void shouldRecycleStateFromStandbyTaskPromotedToActiveTaskAndNotRestore(final boolean useNewProtocol) throws Exception {
         final StreamsBuilder builder = new StreamsBuilder();
         builder.table(
                 inputStream,
@@ -576,7 +544,7 @@ public void shouldRecycleStateFromStandbyTaskPromotedToActiveTaskAndNotRestore(f
             CLUSTER.setGroupStandbyReplicas(appId, 1);
         }
 
-        final Properties props1 = props(stateUpdaterEnabled);
+        final Properties props1 = props();
         props1.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
         props1.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId + "-1").getPath());
         if (useNewProtocol) {
@@ -585,7 +553,7 @@ public void shouldRecycleStateFromStandbyTaskPromotedToActiveTaskAndNotRestore(f
         purgeLocalStreamsState(props1);
         final KafkaStreams streams1 = new KafkaStreams(builder.build(), props1);
 
-        final Properties props2 = props(stateUpdaterEnabled);
+        final Properties props2 = props();
         props2.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
         props2.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId + "-2").getPath());
         if (useNewProtocol) {