wcarlson5 commented on a change in pull request #9697:
URL: https://github.com/apache/kafka/pull/9697#discussion_r540308752
##########
File path: 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();
Review comment:
I don't think it matters, it just set the thread state earlier, but we
can delete it
##########
File path: 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);
+ }
Review comment:
we can do that
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -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
Review comment:
I edited the test to be based on the number of thread instead of hard
coding. And gave a reason
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -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;
+ });
Review comment:
We can change the test so that we verify the replaced threads can
process records.
I am not sure that is necessary for the shutdown as testing if streams can
process some records once started should be tested elsewhere, but in any case I
think that the PR is not the place for this discussion
----------------------------------------------------------------
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
