KAFKA-19666: Clean up integration tests related to state-updater (#20462)

Clean up `KafkaStreamsTelemetryIntegrationTest.java`

Reviewers: Lucas Brutschy <[email protected]>

Author: shashankhs11

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

@@ -65,6 +65,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -174,7 +175,7 @@ public void tearDown() throws Exception {
     @ParameterizedTest
     @MethodSource("recordingLevelParameters")
     public void shouldPushGlobalThreadMetricsToBroker(final String recordingLevel, final String groupProtocol) throws Exception {
-        streamsApplicationProperties = props(true, groupProtocol);
+        streamsApplicationProperties = props(groupProtocol);
         streamsApplicationProperties.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, recordingLevel);
         final Topology topology = simpleTopology(true);
         subscribeForStreamsMetrics();
@@ -213,7 +214,7 @@ public void shouldPushGlobalThreadMetricsToBroker(final String recordingLevel, f
     @MethodSource("recordingLevelParameters")
     public void shouldPushMetricsToBroker(final String recordingLevel, final String groupProtocol) throws Exception {
         // End-to-end test validating metrics pushed to broker
-        streamsApplicationProperties  = props(true, groupProtocol);
+        streamsApplicationProperties  = props(groupProtocol);
         streamsApplicationProperties.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, recordingLevel);
         final Topology topology = simpleTopology(false);
         subscribeForStreamsMetrics();
@@ -274,10 +275,10 @@ public void shouldPushMetricsToBroker(final String recordingLevel, final String
     }
 
     @ParameterizedTest
-    @MethodSource("singleAndMultiTaskParameters")
-    public void shouldPassMetrics(final String topologyType, final boolean stateUpdaterEnabled, final String groupProtocol) throws Exception {
+    @MethodSource("topologyComplexityAndRebalanceProtocol")
+    public void shouldPassMetrics(final String topologyType, final String groupProtocol) throws Exception {
         // Streams metrics should get passed to Admin and Consumer
-        streamsApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
+        streamsApplicationProperties = props(groupProtocol);
         final Topology topology = topologyType.equals("simple") ? simpleTopology(false) : complexTopology();
 
         try (final KafkaStreams streams = new KafkaStreams(topology, streamsApplicationProperties)) {
@@ -303,16 +304,15 @@ public void shouldPassMetrics(final String topologyType, final boolean stateUpda
         }
     }
 
-    @ParameterizedTest
-    @MethodSource("multiTaskParameters")
-    public void shouldPassCorrectMetricsDynamicInstances(final boolean stateUpdaterEnabled, final String groupProtocol) throws Exception {
+    @Test
+    public void shouldPassCorrectMetricsDynamicInstances() throws Exception {
         // Correct streams metrics should get passed with dynamic membership
-        streamsApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
+        streamsApplicationProperties = props("classic");
         streamsApplicationProperties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).getPath() + "-ks1");
         streamsApplicationProperties.put(StreamsConfig.CLIENT_ID_CONFIG, appId + "-ks1");
 
 
-        streamsSecondApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
+        streamsSecondApplicationProperties = props("classic");
         streamsSecondApplicationProperties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).getPath() + "-ks2");
         streamsSecondApplicationProperties.put(StreamsConfig.CLIENT_ID_CONFIG, appId + "-ks2");
 
@@ -407,7 +407,7 @@ public void shouldPassCorrectMetricsDynamicInstances(final boolean stateUpdaterE
     @ValueSource(strings = {"classic", "streams"})
     public void passedMetricsShouldNotLeakIntoClientMetrics(final String groupProtocol) throws Exception {
         // Streams metrics should not be visible in client metrics
-        streamsApplicationProperties = props(true, groupProtocol);
+        streamsApplicationProperties = props(groupProtocol);
         final Topology topology =  complexTopology();
 
         try (final KafkaStreams streams = new KafkaStreams(topology, streamsApplicationProperties)) {
@@ -444,27 +444,16 @@ private List<String> getTaskIdsAsStrings(final KafkaStreams streams) {
                 .toList();
     }
 
-    private static Stream<Arguments> singleAndMultiTaskParameters() {
-        return Stream.of(
-            Arguments.of("simple", true, "classic"),
-            Arguments.of("simple", false, "classic"),
-            Arguments.of("complex", true, "classic"),
-            Arguments.of("complex", false, "classic"),
-            Arguments.of("simple", true, "streams"),
-            Arguments.of("simple", false, "streams")
-        );
-    }
-
-    private static Stream<Arguments> multiTaskParameters() {
+    private static Stream<Arguments> topologyComplexityAndRebalanceProtocol() {
         return Stream.of(
-            Arguments.of(true, "classic"),
-            Arguments.of(false, "classic")
+            Arguments.of("simple", "classic"),
+            Arguments.of("complex", "classic"),
+            Arguments.of("simple", "streams")
         );
     }
 
-    private Properties props(final boolean stateUpdaterEnabled, final String groupProtocol) {
+    private Properties props(final String groupProtocol) {
         return props(mkObjectProperties(mkMap(
-            mkEntry(StreamsConfig.InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled),
             mkEntry(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)
         )));
     }