cadonna commented on a change in pull request #9697:
URL: https://github.com/apache/kafka/pull/9697#discussion_r540865043
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -436,6 +436,26 @@ private void defaultStreamsUncaughtExceptionHandler(final
Throwable throwable) {
}
}
+ private void replaceThreadHelper(final Throwable throwable) {
Review comment:
Please rename to `replaceStreamThread()`.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java
##########
@@ -27,6 +27,7 @@
* Enumeration that describes the response from the exception handler.
*/
enum StreamThreadExceptionResponse {
+ REPLACE_THREAD(0, "REPLACE_STREAM_THREAD"),
Review comment:
I would rename it to `REPLACE_STREAM_THREAD`.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -155,16 +171,46 @@ public void shouldShutdownSingleThreadApplication()
throws InterruptedException
testShutdownApplication(1);
}
+ @Test
+ public void testGlobalThreadException() throws InterruptedException {
+ builder = new StreamsBuilder();
+ builder.addGlobalStore(
+ new KeyValueStoreBuilder<>(
+ Stores.persistentKeyValueStore("globalStore"),
+ Serdes.String(),
+ Serdes.String(),
+ CLUSTER.time
+ ),
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ () -> new ShutdownProcessor(processorValueCollector)
+ );
+ properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 0);
+
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+ kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should
not hit old handler"));
+ kafkaStreams.setUncaughtExceptionHandler(exception ->
SHUTDOWN_CLIENT);
Review comment:
The point of this test should be that a global stream thread is not
replaced but the client is shutdown instead. Hence, the uncaught exception
handler should return `REPLACE_THREAD`, not `SHUTDOWN_CLIENT`.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -155,16 +171,46 @@ public void shouldShutdownSingleThreadApplication()
throws InterruptedException
testShutdownApplication(1);
}
+ @Test
+ public void testGlobalThreadException() throws InterruptedException {
Review comment:
Please rename to something like
`shouldShutDownClientIfGlobalStreamThreadWantsToReplaceThread()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
