KAFKA-19118: Enable KIP-1071 in InternalTopicIntegrationTest (#19425)

KIP-1071 creates internal topics broker-side, so this test checks
whether, when KIP-1071 is enabled, basically the same topics are
created.

It also adds a little helper method in `EmbeddedKafkaCluster`, so that
fewer code changes are required to enable KIP-1071. We use that helper
in the already enabled SmokeTestDriverIntegrationTest and revert some of
the changes there (making the cluster `final` again).

Reviewers: Bill Bejeck <[email protected]>, PoAn Yang
 <[email protected]>

Author: lucasbru

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

@@ -28,6 +28,7 @@
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.streams.GroupProtocol;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
@@ -48,8 +49,9 @@
 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.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -73,7 +75,7 @@
 @Timeout(600)
 @Tag("integration")
 public class InternalTopicIntegrationTest {
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+    public static final EmbeddedKafkaCluster CLUSTER = EmbeddedKafkaCluster.withStreamsRebalanceProtocol(1);
 
     @BeforeAll
     public static void startCluster() throws IOException, InterruptedException {
@@ -147,14 +149,22 @@ private Admin createAdminClient() {
         return Admin.create(adminClientConfig);
     }
 
+    private void configureStreams(final boolean streamsProtocolEnabled, final String appID) {
+        streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+        if (streamsProtocolEnabled) {
+            streamsProp.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+        }
+    }
+
     /*
      * This test just ensures that the assignor does not get stuck during partition number resolution
      * for internal repartition topics. See KAFKA-10689
      */
-    @Test
-    public void shouldGetToRunningWithWindowedTableInFKJ() throws Exception {
-        final String appID = APP_ID + "-windowed-FKJ";
-        streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void shouldGetToRunningWithWindowedTableInFKJ(final boolean streamsProtocolEnabled) throws Exception {
+        final String appID = APP_ID + "-windowed-FKJ-" + streamsProtocolEnabled;
+        configureStreams(streamsProtocolEnabled, appID);
 
         final StreamsBuilder streamsBuilder = new StreamsBuilder();
         final KStream<String, String> inputTopic = streamsBuilder.stream(DEFAULT_INPUT_TOPIC);
@@ -179,10 +189,12 @@ public void shouldGetToRunningWithWindowedTableInFKJ() throws Exception {
         }
     }
 
-    @Test
-    public void shouldCompactTopicsForKeyValueStoreChangelogs() throws Exception {
-        final String appID = APP_ID + "-compact";
-        streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void shouldCompactTopicsForKeyValueStoreChangelogs(final boolean streamsProtocolEnabled) throws Exception {
+        final String appID = APP_ID + "-compact-" + streamsProtocolEnabled;
+        configureStreams(streamsProtocolEnabled, appID);
 
         //
         // Step 1: Configure and start a simple word count topology
@@ -216,10 +228,11 @@ public void shouldCompactTopicsForKeyValueStoreChangelogs() throws Exception {
         assertEquals(4, repartitionProps.size());
     }
 
-    @Test
-    public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs() throws Exception {
-        final String appID = APP_ID + "-compact-delete";
-        streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs(final boolean streamsProtocolEnabled) throws Exception {
+        final String appID = APP_ID + "-compact-delete-" + streamsProtocolEnabled;
+        configureStreams(streamsProtocolEnabled, appID);
 
         //
         // Step 1: Configure and start a simple word count topology

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

@@ -18,8 +18,6 @@
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.utils.Exit;
-import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
-import org.apache.kafka.server.config.ServerConfigs;
 import org.apache.kafka.streams.GroupProtocol;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.StreamsConfig.InternalConfig;
@@ -53,21 +51,17 @@
 @Timeout(600)
 @Tag("integration")
 public class SmokeTestDriverIntegrationTest {
-    public static EmbeddedKafkaCluster cluster;
+    public static final EmbeddedKafkaCluster CLUSTER = EmbeddedKafkaCluster.withStreamsRebalanceProtocol(3);
     public TestInfo testInfo;
 
     @BeforeAll
     public static void startCluster() throws IOException {
-        final Properties props = new Properties();
-        props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
-        props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
-        cluster = new EmbeddedKafkaCluster(3, props);
-        cluster.start();
+        CLUSTER.start();
     }
 
     @AfterAll
     public static void closeCluster() {
-        cluster.stop();
+        CLUSTER.stop();
     }
 
     @BeforeEach
@@ -135,9 +129,9 @@ public void shouldWorkWithRebalance(
         int numClientsCreated = 0;
         final ArrayList<SmokeTestClient> clients = new ArrayList<>();
 
-        IntegrationTestUtils.cleanStateBeforeTest(cluster, SmokeTestDriver.topics());
+        IntegrationTestUtils.cleanStateBeforeTest(CLUSTER, SmokeTestDriver.topics());
 
-        final String bootstrapServers = cluster.bootstrapServers();
+        final String bootstrapServers = CLUSTER.bootstrapServers();
         final Driver driver = new Driver(bootstrapServers, 10, 1000);
         driver.start();
         System.out.println("started driver");

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

@@ -140,6 +140,16 @@ public EmbeddedKafkaCluster(final int numBrokers,
         this.time = new MockTime(mockTimeMillisStart, mockTimeNanoStart);
     }
 
+    public static EmbeddedKafkaCluster withStreamsRebalanceProtocol(final int numBrokers) {
+        return withStreamsRebalanceProtocol(numBrokers, new Properties());
+    }
+
+    public static EmbeddedKafkaCluster withStreamsRebalanceProtocol(final int numBrokers, final Properties props) {
+        props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
+        props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
+        return new EmbeddedKafkaCluster(numBrokers, props);
+    }
+
     public void start() {
         try {
             cluster.format();