cadonna commented on a change in pull request #9487:
URL: https://github.com/apache/kafka/pull/9487#discussion_r523028163
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -366,6 +375,93 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when an {@link
StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread}
+ * throws an unexpected exception.
+ * These might be exceptions indicating rare bugs in Kafka Streams, or they
+ * might be exceptions thrown by your code, for example a
NullPointerException thrown from your processor
+ * logic.
+ * The handler will execute on the thread that produced the exception.
+ * So inorder to get the thread as the java handler type uses use
Thread.currentThread()
+ * <p>
+ * Note, this handler must be threadsafe, since it will be shared among
all threads, and invoked from any
+ * thread that encounters such an exception.
+ *
+ * @param streamsUncaughtExceptionHandler the uncaught exception handler
of type {@link StreamsUncaughtExceptionHandler} for all internal threads
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ * @throws NullPointerException if streamsUncaughtExceptionHandler is null.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final Consumer<Throwable> handler = exception ->
handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ Objects.requireNonNull(streamsUncaughtExceptionHandler);
+ for (final StreamThread thread : threads) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ }
+ if (globalStreamThread != null) {
+ globalStreamThread.setUncaughtExceptionHandler(handler);
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
Review comment:
nit: usually we indent 4 spaces, not 8.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -366,6 +375,93 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when an {@link
StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread}
+ * throws an unexpected exception.
+ * These might be exceptions indicating rare bugs in Kafka Streams, or they
+ * might be exceptions thrown by your code, for example a
NullPointerException thrown from your processor
+ * logic.
+ * The handler will execute on the thread that produced the exception.
+ * So inorder to get the thread as the java handler type uses use
Thread.currentThread()
+ * <p>
+ * Note, this handler must be threadsafe, since it will be shared among
all threads, and invoked from any
+ * thread that encounters such an exception.
+ *
+ * @param streamsUncaughtExceptionHandler the uncaught exception handler
of type {@link StreamsUncaughtExceptionHandler} for all internal threads
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ * @throws NullPointerException if streamsUncaughtExceptionHandler is null.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final Consumer<Throwable> handler = exception ->
handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ Objects.requireNonNull(streamsUncaughtExceptionHandler);
+ for (final StreamThread thread : threads) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ }
+ if (globalStreamThread != null) {
+ globalStreamThread.setUncaughtExceptionHandler(handler);
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
+ }
+ }
+ }
+
+ private void handleStreamsUncaughtExceptionDefaultWrapper(final Throwable
throwable) {
+ if (Thread.getDefaultUncaughtExceptionHandler() != null) {
+ 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);
+ }
+ } else {
+ handleStreamsUncaughtException(throwable, t -> SHUTDOWN_CLIENT);
+ }
+ }
+
+ private void handleStreamsUncaughtException(final Throwable e,
+ final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
action = streamsUncaughtExceptionHandler.handle(e);
+ if (Thread.getDefaultUncaughtExceptionHandler() != null) {
+ 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 SHUTDOWN_CLIENT:
+ log.error("Encountered the following exception during
processing " +
+ "and the registered exception handler opted to \" +
action + \"." +
Review comment:
I guess, you wanted to do this
```suggestion
"and the registered exception handler opted to " +
action + "." +
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -366,6 +375,93 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when an {@link
StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread}
+ * throws an unexpected exception.
+ * These might be exceptions indicating rare bugs in Kafka Streams, or they
+ * might be exceptions thrown by your code, for example a
NullPointerException thrown from your processor
+ * logic.
+ * The handler will execute on the thread that produced the exception.
+ * So inorder to get the thread as the java handler type uses use
Thread.currentThread()
+ * <p>
+ * Note, this handler must be threadsafe, since it will be shared among
all threads, and invoked from any
+ * thread that encounters such an exception.
+ *
+ * @param streamsUncaughtExceptionHandler the uncaught exception handler
of type {@link StreamsUncaughtExceptionHandler} for all internal threads
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ * @throws NullPointerException if streamsUncaughtExceptionHandler is null.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final Consumer<Throwable> handler = exception ->
handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ Objects.requireNonNull(streamsUncaughtExceptionHandler);
+ for (final StreamThread thread : threads) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ }
+ if (globalStreamThread != null) {
+ globalStreamThread.setUncaughtExceptionHandler(handler);
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
+ }
+ }
+ }
+
+ private void handleStreamsUncaughtExceptionDefaultWrapper(final Throwable
throwable) {
+ if (Thread.getDefaultUncaughtExceptionHandler() != null) {
+ 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);
+ }
+ } else {
+ handleStreamsUncaughtException(throwable, t -> SHUTDOWN_CLIENT);
+ }
+ }
+
+ private void handleStreamsUncaughtException(final Throwable e,
+ final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
action = streamsUncaughtExceptionHandler.handle(e);
+ if (Thread.getDefaultUncaughtExceptionHandler() != null) {
Review comment:
Are you sure this is the correct method to call? As far as I understand
the the javadocs and the decompiled code, this method does not return the
handler you can set on a `Thread` with `setUncaughtExceptionHandler()`.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -366,6 +374,63 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when an {@link
StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread}
+ * throws an unexpected exception.
+ * These might be exceptions indicating rare bugs in Kafka Streams, or they
+ * might be exceptions thrown by your code, for example a
NullPointerException thrown from your processor
+ * logic.
+ * <p>
+ * Note, this handler must be threadsafe, since it will be shared among
all threads, and invoked from any
+ * thread that encounters such an exception.
+ *
+ * @param streamsUncaughtExceptionHandler the uncaught exception handler
of type {@link StreamsUncaughtExceptionHandler} for all internal threads
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ * @throws NullPointerException if streamsUncaughtExceptionHandler is null.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler handler = exception ->
handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ Objects.requireNonNull(streamsUncaughtExceptionHandler);
+ for (final StreamThread thread : threads) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ }
+ if (globalStreamThread != null) {
+ globalStreamThread.setUncaughtExceptionHandler(handler);
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
+ }
+ }
+ }
+
+ private StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
handleStreamsUncaughtException(final Throwable e,
+
final StreamsUncaughtExceptionHandler
streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
action = streamsUncaughtExceptionHandler.handle(e);
+ switch (action) {
+ case SHUTDOWN_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. ", e);
+ close(Duration.ZERO);
Review comment:
I still have a question here. Since the stream thread is alive when it
calls `close()` there will not be a deadlock anymore. So, why do we call
`close()` with duration zero?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -366,6 +375,93 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when an {@link
StreamsConfig#NUM_STREAM_THREADS_CONFIG internal thread}
+ * throws an unexpected exception.
+ * These might be exceptions indicating rare bugs in Kafka Streams, or they
+ * might be exceptions thrown by your code, for example a
NullPointerException thrown from your processor
+ * logic.
+ * The handler will execute on the thread that produced the exception.
+ * So inorder to get the thread as the java handler type uses use
Thread.currentThread()
+ * <p>
+ * Note, this handler must be threadsafe, since it will be shared among
all threads, and invoked from any
+ * thread that encounters such an exception.
+ *
+ * @param streamsUncaughtExceptionHandler the uncaught exception handler
of type {@link StreamsUncaughtExceptionHandler} for all internal threads
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ * @throws NullPointerException if streamsUncaughtExceptionHandler is null.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final Consumer<Throwable> handler = exception ->
handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ Objects.requireNonNull(streamsUncaughtExceptionHandler);
+ for (final StreamThread thread : threads) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ }
+ if (globalStreamThread != null) {
+ globalStreamThread.setUncaughtExceptionHandler(handler);
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
+ }
+ }
+ }
+
+ private void handleStreamsUncaughtExceptionDefaultWrapper(final Throwable
throwable) {
+ if (Thread.getDefaultUncaughtExceptionHandler() != null) {
+ 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);
+ }
+ } else {
+ handleStreamsUncaughtException(throwable, t -> SHUTDOWN_CLIENT);
+ }
+ }
+
+ private void handleStreamsUncaughtException(final Throwable e,
Review comment:
Please use a more meaningful parameter name.
----------------------------------------------------------------
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
