MINOR: Deflake and improve SmokeTestDriverIntegrationTest (#20509)

This improves the SmokeTestDriverIntegrationTest in three ways:

1) If a SmokeTestClient fails (enters a terminal ERROR state), the
SmokeTestDriverIntegrationTest currently times out, because it keeps
waiting for state NOT_RUNNING. This makes debugging quite difficult.
This minor  change makes sure to just fail the test immediately, if a
SmokeTestClient enters the ERROR state.

2) If a test times out or fails prematurely, because a SmokeTestClient
crashed, the SmokeTestClients aren't shut down correctly, which will
affect the following test runs. Therefore, I am adding clean-up logic
that running SmokeTestClients in `@AfterAll`.

3) Finally, I found that the processingThread variation of this thread
triggers a subtle race condition. Since this features is currently not
actively developed, I disabled those variations and created a ticket to
reactivate the test.

Reviewers: Matthias J. Sax <[email protected]>, Chia-Ping Tsai
 <[email protected]>, Bill Bejeck <[email protected]>

---------

Co-authored-by: Copilot <[email protected]>

Author: lucasbru

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

@@ -27,6 +27,7 @@
 import org.apache.kafka.streams.tests.SmokeTestDriver;
 
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
@@ -46,29 +47,43 @@
 import static org.apache.kafka.streams.tests.SmokeTestDriver.generate;
 import static org.apache.kafka.streams.tests.SmokeTestDriver.verify;
 import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Timeout(600)
 @Tag("integration")
 public class SmokeTestDriverIntegrationTest {
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);
+    private static EmbeddedKafkaCluster cluster = null;
     public TestInfo testInfo;
+    private ArrayList<SmokeTestClient> clients = new ArrayList<>();
 
     @BeforeAll
     public static void startCluster() throws IOException {
-        CLUSTER.start();
+        cluster = new EmbeddedKafkaCluster(3);
+        cluster.start();
     }
 
     @AfterAll
     public static void closeCluster() {
-        CLUSTER.stop();
+        cluster.stop();
+        cluster = null;
     }
 
     @BeforeEach
     public void setUp(final TestInfo testInfo) {
         this.testInfo = testInfo;
     }
 
+    @AfterEach
+    public void shutDown(final TestInfo testInfo) {
+        // Clean up clients in case the test failed or timed out
+        for (final SmokeTestClient client : clients) {
+            if (!client.closed() && !client.error()) {
+                client.close();
+            }
+        }
+    }
+
     private static class Driver extends Thread {
         private final String bootstrapServers;
         private final int numKeys;
@@ -108,12 +123,11 @@ SmokeTestDriver.VerificationResult result() {
     // During the new stream added and old stream left, the stream process should still complete without issue.
     // We set 2 timeout condition to fail the test before passing the verification:
     // (1) 10 min timeout, (2) 30 tries of polling without getting any data
+    // The processing thread variations where disabled since they triggered a race condition, see KAFKA-19696
     @ParameterizedTest
     @CsvSource({
         "false, true",
-        "false, false",
-        "true, true",
-        "true, false"
+        "false, false"
     })
     public void shouldWorkWithRebalance(
         final boolean processingThreadsEnabled,
@@ -126,11 +140,10 @@ public void shouldWorkWithRebalance(
             throw new AssertionError("Test called halt(). code:" + statusCode + " message:" + message);
         });
         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");
@@ -144,8 +157,8 @@ public void shouldWorkWithRebalance(
         if (streamsProtocolEnabled) {
             props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
             // decrease the session timeout so that we can trigger the rebalance soon after old client left closed
-            CLUSTER.setGroupSessionTimeout(appId, 10000);
-            CLUSTER.setGroupHeartbeatTimeout(appId, 1000);
+            cluster.setGroupSessionTimeout(appId, 10000);
+            cluster.setGroupHeartbeatTimeout(appId, 1000);
         } else {
             // decrease the session timeout so that we can trigger the rebalance soon after old client left closed
             props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
@@ -167,6 +180,7 @@ public void shouldWorkWithRebalance(
 
                 client.closeAsync();
                 while (!client.closed()) {
+                    assertFalse(client.error(), "The streams application seems to have crashed.");
                     Thread.sleep(100);
                 }
             }
@@ -184,6 +198,7 @@ public void shouldWorkWithRebalance(
             // then, wait for them to stop
             for (final SmokeTestClient client : clients) {
                 while (!client.closed()) {
+                    assertFalse(client.error(), "The streams application seems to have crashed.");
                     Thread.sleep(100);
                 }
             }

streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java

@@ -54,6 +54,7 @@ public class SmokeTestClient extends SmokeTestUtil {
     private KafkaStreams streams;
     private boolean uncaughtException = false;
     private volatile boolean closed;
+    private volatile boolean error;
 
     private static void addShutdownHook(final String name, final Runnable runnable) {
         if (name != null) {
@@ -71,6 +72,10 @@ public boolean closed() {
         return closed;
     }
 
+    public boolean error() {
+        return error;
+    }
+
     public void start(final Properties streamsProperties) {
         final Topology build = getTopology();
         streams = new KafkaStreams(build, getStreamsConfig(streamsProperties));
@@ -85,6 +90,10 @@ public void start(final Properties streamsProperties) {
             if (newState == KafkaStreams.State.NOT_RUNNING) {
                 closed = true;
             }
+
+            if (newState == KafkaStreams.State.ERROR) {
+                error = true;
+            }
         });
 
         streams.setUncaughtExceptionHandler(e -> {