mjsax commented on code in PR #23093:
URL: https://github.com/apache/kafka/pull/23093#discussion_r3726137022
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -862,6 +858,7 @@ public boolean checkStateUpdater(final long now,
if (stateUpdater.hasExceptionsAndFailedTasks()) {
handleExceptionsFromStateUpdater();
}
+ maybeThrowFatalExceptionFromStateUpdater();
Review Comment:
What is the reason we insert this check exactly here, but not at the very
beginning or very end?
If the state-updater is already dead, why would we want to still add tasks?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -167,19 +169,28 @@ public void run() {
while (isRunning.get()) {
runOnce();
}
- } catch (final RuntimeException anyOtherException) {
- handleRuntimeException(anyOtherException);
+ } catch (final Throwable anyOtherThrowable) {
+ handleFatalThrowable(anyOtherThrowable);
} finally {
- clearInputQueue();
- clearUpdatingAndPausedTasks();
+ failRemainingTasks();
+ failPendingActions();
Review Comment:
We are changing the order here -- is this an intended change?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -338,9 +351,18 @@ private KafkaFutureImpl<Uuid>
restoreConsumerInstanceId(final Duration timeout)
}
- private void handleRuntimeException(final RuntimeException
runtimeException) {
- log.error("An unexpected error occurred within the state updater
thread: {}", String.valueOf(runtimeException));
-
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(runtimeException);
+ // Any throwable that escapes the update loop stops this thread. Even
for a throwable that is not recoverable,
+ // like an error, we report the tasks of this thread as failed, so
that the stream thread notices that this
+ // thread is gone instead of silently losing its tasks.
+ private void handleFatalThrowable(final Throwable throwable) {
+ log.error("An unexpected error occurred within the state updater
thread: {}", String.valueOf(throwable));
+ final RuntimeException exception = throwable instanceof
RuntimeException
+ ? (RuntimeException) throwable
+ : new StreamsException("The state updater thread failed with a
fatal error.", throwable);
Review Comment:
Not sure why we need this ? Both a general `RuntimeException`,
`StreamsException`, and `Throwable` are all fatal. So why these different cases?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -580,9 +620,6 @@ private void removeTask(final TaskId taskId,
} catch (final StreamsException streamsException) {
handleStreamsExceptionWithTask(streamsException, taskId);
future.completeExceptionally(streamsException);
- } catch (final RuntimeException runtimeException) {
Review Comment:
Why do we keep `StreamsException` as-is but only remove `RuntimeException`?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -167,19 +169,28 @@ public void run() {
while (isRunning.get()) {
runOnce();
}
- } catch (final RuntimeException anyOtherException) {
- handleRuntimeException(anyOtherException);
+ } catch (final Throwable anyOtherThrowable) {
+ handleFatalThrowable(anyOtherThrowable);
Review Comment:
It seems we should still re-throw the exception?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -338,9 +351,18 @@ private KafkaFutureImpl<Uuid>
restoreConsumerInstanceId(final Duration timeout)
}
- private void handleRuntimeException(final RuntimeException
runtimeException) {
- log.error("An unexpected error occurred within the state updater
thread: {}", String.valueOf(runtimeException));
-
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(runtimeException);
+ // Any throwable that escapes the update loop stops this thread. Even
for a throwable that is not recoverable,
+ // like an error, we report the tasks of this thread as failed, so
that the stream thread notices that this
+ // thread is gone instead of silently losing its tasks.
+ private void handleFatalThrowable(final Throwable throwable) {
+ log.error("An unexpected error occurred within the state updater
thread: {}", String.valueOf(throwable));
Review Comment:
Why do we not log the proper stacktrace but only `String.valueOf(throwable)`?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
