wcarlson5 commented on a change in pull request #9273:
URL: https://github.com/apache/kafka/pull/9273#discussion_r501124504
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -364,6 +370,73 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when a {@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;
{@code null} deletes the current handler
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler handler = throwable ->
handleStreamsUncaughtException(throwable, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ for (final StreamThread thread : threads) {
Review comment:
I am not sure, we never talked about the global stream thread.
I don't think that it makes sense to use the same handler as its is quite
different than a normal streamThread. I would lean towards making a separate
method to just set it like it was
##########
File path:
streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+public interface StreamsUncaughtExceptionHandler {
+ /**
+ * Inspect a record and the exception received.
+ * @param exception the actual exception
+ */
+ StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse
handle(final Throwable exception);
+
+ /**
+ * Enumeration that describes the response from the exception handler.
+ */
+ enum StreamsUncaughtExceptionHandlerResponse {
+
+
Review comment:
okay
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -566,10 +572,36 @@ void runLoop() {
}
} catch (final TaskMigratedException e) {
handleTaskMigrated(e);
+ } catch (final Exception e) {
+ if (streamsUncaughtExceptionHandler != null) {
Review comment:
this check is redundant. It should never be null here, I will remove it
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -936,6 +968,8 @@ public void shutdown() {
}
}
+
+
Review comment:
removed
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -364,6 +370,73 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when a {@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;
{@code null} deletes the current handler
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler handler = throwable ->
handleStreamsUncaughtException(throwable, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ for (final StreamThread thread : threads) {
+ if (streamsUncaughtExceptionHandler != null) {
Review comment:
I am not so sure. The null in the other handler is used to "reset" to
the default behavior. I already set in the constructor
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -528,12 +536,10 @@ public void run() {
throw e;
}
}
-
- log.error("Encountered the following exception during processing "
+
- "and the thread is going to shut down: ", e);
throw e;
} finally {
completeShutdown(cleanRun);
+
Review comment:
removed
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -436,6 +496,8 @@ private void maybeSetError() {
}
if (setState(State.ERROR)) {
+ metrics.close();
Review comment:
That does make sense, I will change it
##########
File path:
streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+public interface StreamsUncaughtExceptionHandler {
+ /**
+ * Inspect a record and the exception received.
+ * @param exception the actual exception
+ */
+ StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse
handle(final Throwable exception);
Review comment:
Since the handler is now executing on the thread we don't need to pass
that in
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -384,9 +390,8 @@ public static StreamThread create(final
InternalTopologyBuilder builder,
builder,
threadId,
logContext,
- assignmentErrorCode,
- nextScheduledRebalanceMs
- );
+ nextScheduledRebalanceMs,
+ assignmentErrorCode);
Review comment:
fixed
##########
File path:
streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+public interface StreamsUncaughtExceptionHandler {
+ /**
+ * Inspect a record and the exception received.
+ * @param exception the actual exception
+ */
+ StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse
handle(final Throwable exception);
+
+ /**
+ * Enumeration that describes the response from the exception handler.
+ */
+ enum StreamsUncaughtExceptionHandlerResponse {
+
+
+ SHUTDOWN_STREAM_THREAD(0, "SHUTDOWN_STREAM_THREAD"),
+// REPLACE_STREAM_THREAD(1, "REPLACE_STREAM_THREAD"),
+ SHUTDOWN_KAFKA_STREAMS_CLIENT(2, "SHUTDOWN_KAFKA_STREAMS_CLIENT"),
+ SHUTDOWN_KAFKA_STREAMS_APPLICATION(3,
"SHUTDOWN_KAFKA_STREAMS_APPLICATION");
+
Review comment:
okay
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -342,6 +347,13 @@ public GroupAssignment assign(final Cluster metadata,
final GroupSubscription gr
// the maximum of the depending sub-topologies source topics' number
of partitions
final Map<Integer, TopicsInfo> topicGroups =
taskManager.builder().topicGroups();
+ if (shutdownRequested) {
Review comment:
no. Moved
##########
File path: streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
##########
@@ -617,7 +619,19 @@ public void
shouldThrowExceptionSettingUncaughtExceptionHandlerNotInCreateState(
final KafkaStreams streams = new
KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
try {
- streams.setUncaughtExceptionHandler(null);
+
streams.setUncaughtExceptionHandler((Thread.UncaughtExceptionHandler) null);
+ fail("Should throw IllegalStateException");
+ } catch (final IllegalStateException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void
shouldThrowExceptionSettingStreamsUncaughtExceptionHandlerNotInCreateState() {
Review comment:
I am not sure what you mean by this can you clarify?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -226,8 +226,8 @@ private StreamThread
createStreamThread(@SuppressWarnings("SameParameterValue")
0,
stateDirectory,
new MockStateRestoreListener(),
- threadIdx
- );
+ threadIdx,
+ new AtomicInteger());
Review comment:
turns out the auto refactoring does not work with new params at the end
of methods. fixed.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -342,6 +347,13 @@ public GroupAssignment assign(final Cluster metadata,
final GroupSubscription gr
// the maximum of the depending sub-topologies source topics' number
of partitions
final Map<Integer, TopicsInfo> topicGroups =
taskManager.builder().topicGroups();
+ if (shutdownRequested) {
+ return new GroupAssignment(
+ errorAssignment(clientMetadataMap,
+ AssignorError.SHUTDOWN_REQUESTED.code())
Review comment:
i can
##########
File path: streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
##########
@@ -617,7 +619,19 @@ public void
shouldThrowExceptionSettingUncaughtExceptionHandlerNotInCreateState(
final KafkaStreams streams = new
KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
try {
- streams.setUncaughtExceptionHandler(null);
+
streams.setUncaughtExceptionHandler((Thread.UncaughtExceptionHandler) null);
+ fail("Should throw IllegalStateException");
+ } catch (final IllegalStateException e) {
+ // expected
+ }
+ }
Review comment:
the UncaughtExceptionHandler which is set should not be called in this
case. Do you mean we should verify it is not called?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -364,6 +370,73 @@ public void setUncaughtExceptionHandler(final
Thread.UncaughtExceptionHandler eh
}
}
+ /**
+ * Set the handler invoked when a {@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;
{@code null} deletes the current handler
+ * @throws IllegalStateException if this {@code KafkaStreams} instance is
not in state {@link State#CREATED CREATED}.
+ */
+ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ final StreamsUncaughtExceptionHandler handler = throwable ->
handleStreamsUncaughtException(throwable, streamsUncaughtExceptionHandler);
+ synchronized (stateLock) {
+ if (state == State.CREATED) {
+ for (final StreamThread thread : threads) {
+ if (streamsUncaughtExceptionHandler != null) {
+ thread.setStreamsUncaughtExceptionHandler(handler);
+ } else {
+ final StreamsUncaughtExceptionHandler defaultHandler =
exception ->
+
StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse.SHUTDOWN_STREAM_THREAD;
+
thread.setStreamsUncaughtExceptionHandler(defaultHandler);
+ }
+ }
+ } else {
+ throw new IllegalStateException("Can only set
UncaughtExceptionHandler in CREATED state. " +
+ "Current state is: " + state);
+ }
+ }
+ }
+
+ private
StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse
handleStreamsUncaughtException(final Throwable e,
+
final StreamsUncaughtExceptionHandler
streamsUncaughtExceptionHandler) {
+ final
StreamsUncaughtExceptionHandler.StreamsUncaughtExceptionHandlerResponse action
= streamsUncaughtExceptionHandler.handle(e);
+ switch (action) {
+ case SHUTDOWN_STREAM_THREAD:
+ log.error("Encountered the following exception during
processing " +
+ "and the thread is going to shut down: ", e);
+ break;
Review comment:
I makes more sense to me to have all the cases in the same place. Since
it is the default all the other cases will execute the rest of the logic as
well so the rethrowing of the error not really in this case.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -932,9 +929,8 @@ public void shouldShutdownTaskManagerOnClose() {
internalTopologyBuilder,
CLIENT_ID,
new LogContext(""),
- new AtomicInteger(),
- new AtomicLong(Long.MAX_VALUE)
- ).updateThreadMetadata(getSharedAdminClientId(CLIENT_ID));
+ new AtomicLong(Long.MAX_VALUE),
+ new
AtomicInteger()).updateThreadMetadata(getSharedAdminClientId(CLIENT_ID));
Review comment:
foxed the 5 of these as well
----------------------------------------------------------------
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
