KAFKA-19724: Global stream thread should not ignore any exceptions (#20668)

Kafka Streams does not catch Error types that occur during
`GlobalStreamThread` initiation, and therefore it is not possible to
trace the error (for example, an `ExceptionInInitializerError` occurs
when RocksDB is not found for a global store). This is because errors
are not caught and logged.

The catch block in `GlobalStreamThread#initialize()` has been ensured to
catch `Throwable` instead of `Exception`. Additionally, the empty
`setUncaughtHandler` set operation that prevented this from taking
effect when users employed setUncaughtExceptionHandler has been removed.

Reviewers: Matthias J. Sax <[email protected]>

Author: celikfatih

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

@@ -476,11 +476,6 @@ public void setUncaughtExceptionHandler(final StreamsUncaughtExceptionHandler us
                 }
                 processStreamThread(thread -> thread.setUncaughtExceptionHandler((t, e) -> { }
                 ));
-
-                if (globalStreamThread != null) {
-                    globalStreamThread.setUncaughtExceptionHandler((t, e) -> { }
-                    );
-                }
             } else {
                 throw new IllegalStateException("Can only set UncaughtExceptionHandler before calling start(). " +
                     "Current state is: " + state);

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

@@ -243,7 +243,7 @@ static class StateConsumer {
         }
 
         /**
-         * @throws IllegalStateException If store gets registered after initialized is already finished
+         * @throws IllegalStateException If a store gets registered after initialized is already finished
          * @throws StreamsException      if the store's change log does not contain the partition
          */
         void initialize() {
@@ -433,7 +433,7 @@ private StateConsumer initialize() {
         } catch (final StreamsException fatalException) {
             closeStateConsumer(stateConsumer, false);
             startupException = fatalException;
-        } catch (final Exception fatalException) {
+        } catch (final Throwable fatalException) {
             closeStateConsumer(stateConsumer, false);
             startupException = new StreamsException("Exception caught during initialization of GlobalStreamThread", fatalException);
         }

streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java

@@ -121,12 +121,7 @@ public void process(final Record<Object, Object> record) {
         );
 
         baseDirectoryName = TestUtils.tempDirectory().getAbsolutePath();
-        final HashMap<String, Object> properties = new HashMap<>();
-        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "blah");
-        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "testAppId");
-        properties.put(StreamsConfig.STATE_DIR_CONFIG, baseDirectoryName);
-        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
-        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
+        final HashMap<String, Object> properties = getStreamProperties();
         config = new StreamsConfig(properties);
         globalStreamThread = new GlobalStreamThread(
             builder.rewriteTopology(config).buildGlobalStateTopology(),
@@ -406,6 +401,51 @@ public void shouldTimeOutOnGlobalConsumerInstanceId() throws Exception {
         }
     }
 
+    @Test
+    public void shouldThrowStreamsExceptionOnStartupIfThrowableOccurred() throws Exception {
+        final String exceptionMessage = "Throwable occurred!";
+        final MockConsumer<byte[], byte[]> consumer = new MockConsumer<byte[], byte[]>(OffsetResetStrategy.EARLIEST) {
+            @Override
+            public List<PartitionInfo> partitionsFor(final String topic) {
+                throw new ExceptionInInitializerError(exceptionMessage);
+            }
+        };
+        final StateStore globalStore = builder.globalStateStores().get(GLOBAL_STORE_NAME);
+        globalStreamThread = new GlobalStreamThread(
+                builder.buildGlobalStateTopology(),
+                config,
+                consumer,
+                new StateDirectory(config, time, true, false),
+                0,
+                new StreamsMetricsImpl(new Metrics(), "test-client", "processId", time),
+                time,
+                "clientId",
+                stateRestoreListener,
+                e -> { }
+        );
+
+        try {
+            globalStreamThread.start();
+            fail("Should have thrown StreamsException if start up failed");
+        } catch (final StreamsException e) {
+            assertThat(e.getCause(), instanceOf(Throwable.class));
+            assertThat(e.getCause().getMessage(), equalTo(exceptionMessage));
+        }
+        globalStreamThread.join();
+        assertThat(globalStore.isOpen(), is(false));
+        assertFalse(globalStreamThread.stillRunning());
+    }
+
+    private HashMap<String, Object> getStreamProperties() {
+        final HashMap<String, Object> properties = new HashMap<>();
+        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "blah");
+        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "testAppId");
+        properties.put(StreamsConfig.STATE_DIR_CONFIG, baseDirectoryName);
+        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
+        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
+        return properties;
+    }
+
     private void initializeConsumer() {
         mockConsumer.updatePartitions(
             GLOBAL_STORE_TOPIC_NAME,