wcarlson5 commented on a change in pull request #9697:
URL: https://github.com/apache/kafka/pull/9697#discussion_r536416836
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -145,6 +148,26 @@ public void shouldShutdownClient() throws
InterruptedException {
}
}
+ @Test
+ public void shouldReplaceThread() throws InterruptedException {
+ try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
+ kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should
not hit old handler"));
+ AtomicInteger count = new AtomicInteger();
+ kafkaStreams.setUncaughtExceptionHandler(exception -> {
+ count.getAndIncrement();
+ return REPLACE_THREAD;
+ });
+
+
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+
+ produceMessages(0L, inputTopic, "A");
+ TestUtils.waitForCondition(() -> count.get() > 2, DEFAULT_TIMEOUT,
"At least 3 threads have died");
Review comment:
All of the original threads have died and been replaced and it is still
running
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -444,6 +444,18 @@ 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:
+ StreamThread deadThread = (StreamThread)
threads.stream().filter(n ->
n.getName().equals(Thread.currentThread().getName())).toArray()[0];
+ threads.remove(deadThread);
Review comment:
removing from the thread list does 2 things
1. keeps DEAD threads out of the list as kip-663 dictates
2. ensures the next thread has the same name
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -444,6 +444,18 @@ 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:
+ StreamThread deadThread = (StreamThread)
threads.stream().filter(n ->
n.getName().equals(Thread.currentThread().getName())).toArray()[0];
+ threads.remove(deadThread);
+ addStreamThread();
+ deadThread.shutdown();
+ if (throwable instanceof RuntimeException) {
Review comment:
We need to throw the error or the task gets lost and we drop records
----------------------------------------------------------------
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
