mjsax commented on code in PR #23093:
URL: https://github.com/apache/kafka/pull/23093#discussion_r3737073660
##########
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:
But what if state-updater is already dead before-hand? Also, why not check
after the removal step completed too?
Claude also has concerns (did not verify the details):
> The current code assumes handleExceptionsFromStateUpdater() returns. It
can't when it has anything to report: maybeThrowTaskExceptions throws on all
three branches (lastFatal, lastTaskMigrated, aggregated TaskCorrupted). And
handleFatalThrowable stamps the fatal exception onto every owned task while
also setting fatalException. So in the scenario the check exists for, line 872
is unreachable; it only fires when the updater died owning zero tasks.
>
> The concrete damage shows up when addTasksToStateUpdater() — which still
runs first, per your follow-up question — has pending tasks in the same
iteration. Those get failAction'd with the generic StreamsException("The state
updater is not running.") and land after the real cause in the drain order.
maybeThrowTaskExceptions keeps the last plain StreamsException as lastFatal, so
the StreamThread dies with "The state updater is not running." and the actual
cause survives only in the state updater's own ERROR log. That directly defeats
the comment Nikita wrote three lines above fatalException.set(exception):
"publish the exception before reporting the tasks as failed, so that the stream
thread never sees a task that failed because this thread died without also
seeing why this thread died."
>
> Moving the check to the top of checkStateUpdater, ahead of
addTasksToStateUpdater(), fixes both that and your "why would we still add
tasks to a dead updater" question in one move.
--
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]
