root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-12699: Override the default handler for stream threads if the stream's handler is used (#12324) 

Override the default handler for stream threads if the stream's handler is used. We do no want the java default handler triggering when a thread is replaced.

Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
This commit is contained in:
Walker Carlson 2022-07-19 15:35:26 -05:00 committed by GitHub
parent 693e283802
commit b62d8b975c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View File

@ -471,6 +471,13 @@ public class KafkaStreams implements AutoCloseable {
exception -> handleStreamsUncaughtException(exception, userStreamsUncaughtExceptionHandler, false)
);
}
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);

6
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
View File

@ -176,6 +176,12 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
testReplaceThreads(2);
}
@Test
public void shouldReplaceThreadsWithoutJavaHandler() throws InterruptedException {
Thread.setDefaultUncaughtExceptionHandler((t, e) -> fail("exception thrown"));
testReplaceThreads(2);
}
@Test
public void shouldReplaceSingleThread() throws InterruptedException {
testReplaceThreads(1);