guozhangwang commented on a change in pull request #11686:
URL: https://github.com/apache/kafka/pull/11686#discussion_r789060082
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -523,16 +528,21 @@ private boolean wrappedExceptionIsIn(final Throwable
throwable, final Set<Class<
}
private void handleStreamsUncaughtException(final Throwable throwable,
- final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler,
+ final boolean
skipThreadReplacement) {
final StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
action = getActionForThrowable(throwable, streamsUncaughtExceptionHandler);
if (oldHandler) {
log.warn("Stream's new uncaught exception handler is set as well
as the deprecated old handler." +
"The old handler will be ignored as long as a new handler
is set.");
}
switch (action) {
case REPLACE_THREAD:
- log.error("Replacing thread in the streams uncaught exception
handler", throwable);
- replaceStreamThread(throwable);
+ if (!skipThreadReplacement) {
+ log.error("Replacing thread in the streams uncaught
exception handler", throwable);
+ replaceStreamThread(throwable);
+ } else {
+ log.debug("Skipping thread replacement for recoverable
error");
Review comment:
+1 on `CONTINUE`, where the library could decide what to do actually,
like whether we should replace the thread or not due to optimizations. I think
it's not too late to replace it via a small KIP --- this does not need to block
this PR though, just that the more I think of it, the more I feel we should
make it more abstract away from implementation details.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
