KAFKA-10810: Replace stream threads

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -444,6 +444,25 @@ private void handleStreamsUncaughtException(final Throwable throwable,
                     "The old handler will be ignored as long as a new handler is set.");
         }
         switch (action) {
+            case REPLACE_THREAD:
+                if (globalStreamThread != null && Thread.currentThread().getName().equals(globalStreamThread.getName())) {
+                    log.warn("The global thread cannot be replaced. Reverting to shutting down the client.");
+                    log.error("Encountered the following exception during processing " +
+                            "and the registered exception handler opted to " + action + "." +
+                            " The streams client is going to shut down now. ", throwable);
+                    close(Duration.ZERO);
+                }
+                final StreamThread deadThread = (StreamThread) Thread.currentThread();
+                threads.remove(deadThread);
+                addStreamThread();
+                deadThread.shutdown();
+                if (throwable instanceof RuntimeException) {
+                    throw (RuntimeException) throwable;
+                } else if (throwable instanceof Error) {
+                    throw (Error) throwable;
+                } else {
+                    throw new RuntimeException("Unexpected checked exception caught in the uncaught exception handler", throwable);
+                }
             case SHUTDOWN_CLIENT:
                 log.error("Encountered the following exception during processing " +
                         "and the registered exception handler opted to " + action + "." +

streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java

@@ -27,6 +27,7 @@ public interface StreamsUncaughtExceptionHandler {
      * Enumeration that describes the response from the exception handler.
      */
     enum StreamThreadExceptionResponse {
+        REPLACE_THREAD(0, "REPLACE_STREAM_THREAD"),
         SHUTDOWN_CLIENT(1, "SHUTDOWN_KAFKA_STREAMS_CLIENT"),
         SHUTDOWN_APPLICATION(2, "SHUTDOWN_KAFKA_STREAMS_APPLICATION");
 

streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java

@@ -53,6 +53,7 @@
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
+import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.purgeLocalStreamsState;
@@ -66,7 +67,7 @@
 @SuppressWarnings("deprecation") //Need to call the old handler, will remove those calls when the old handler is removed
 public class StreamsUncaughtExceptionHandlerIntegrationTest {
     @ClassRule
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1, new Properties(), 0L, 0L);
     public static final Duration DEFAULT_DURATION = Duration.ofSeconds(30);
 
     @Rule
@@ -145,6 +146,16 @@ public void shouldShutdownClient() throws InterruptedException {
         }
     }
 
+    @Test
+    public void shouldReplaceThreads() throws InterruptedException {
+        testReplaceThreads(2);
+    }
+
+    @Test
+    public void shouldReplaceSingleThread() throws InterruptedException {
+        testReplaceThreads(1);
+    }
+
     @Test
     public void shouldShutdownMultipleThreadApplication() throws InterruptedException {
         testShutdownApplication(2);
@@ -202,6 +213,29 @@ private void testShutdownApplication(final int numThreads) throws InterruptedExc
             assertThat(processorValueCollector.size(), equalTo(1));
         }
     }
+
+    private void testReplaceThreads(final int numThreads) throws InterruptedException {
+        properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads);
+        try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
+            kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));
+
+            final AtomicInteger count = new AtomicInteger();
+            kafkaStreams.setUncaughtExceptionHandler(exception -> {
+                count.getAndIncrement();
+                if (count.get() > 2) {
+                    return SHUTDOWN_CLIENT;
+                }
+                return REPLACE_THREAD;
+            });
+            StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+
+            produceMessages(0L, inputTopic, "A");
+            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);
+
+            assertThat(processorValueCollector.size(), equalTo(3));
+            //because we only have 2 threads at the start and each record kills a thread we must have replaced threads
+        }
+    }
 }