KAFKA-20721: Removed timeout on task removal from state updater.

[Nikita-Shupletsov]

• Added logic to catch and propagate to the stream thread fatal state update errors.
• Improved the error handling logic in the state updater thread. Now when the thread is failing it marks all tasks as failed.
• Added logic to fail task removal futures if the thread is dead.

Reviewers: Matthias J. Sax matthias@confluent.io

[mjsax]

Nice fix! Overall this make sense to me.

Few more Claude comments:

future.get() blocks the StreamThread inside the rebalance callback with zero log output — the old code at least printed "The state updater wasn't able to remove task X in time." Suggest looping future.get(1, MINUTES) with a WARN per iteration and retrying forever: keeps the fix, restores observability. Same applies to shutdownStateUpdater(), where KafkaStreams.close(Duration) will now return false and leak the StreamThread instead of the previously-bounded 5-min-per-future.

Integration test coverage regression. StateUpdaterFailureIntegrationTest was added by c48c50d (KAFKA-19831) specifically for "potential failures in Task#maybeCheckpoint". The PR deletes the flush() override that injected a ProcessorStateException during maybeCheckpointTasks and replaces it with a throw from onRestoreEnd, i.e. inside restoreTasks. Different handler — maybeCheckpointTasks catches StreamsException per task and continues; restoreTasks goes through handleStreamsException. The flush case should stay as a third Arguments rather than being replaced.

StateUpdater.shutdown() semantics changed for restored tasks. failRemainingTasks() moves everything in restoredActiveTasks into the failed queue, and shouldShutdownStateUpdater now asserts that. Those tasks restored successfully; being "failed" means closeTaskDirty, discarding the checkpoint. It doesn't bite production today because shutdownStateUpdater() drains them via removeRestoredTask() before calling shutdown() — but it makes shutdown() unsafe to call on its own, which the unit test now enshrines as expected behavior.

[mjsax]

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?

[Nikita-Shupletsov]

it goes after handleExceptionsFromStateUpdater, because it calls drainExceptionsAndFailedTasks and processes them(adds them to the failed tasks). so we first let that logic do its thing, then we check if we need to fail the thread

[mjsax]

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.

[mjsax]

Why do we keep StreamsException as-is but only remove RuntimeException?

[Nikita-Shupletsov]

because we don't stop the thread on StreamsException, but we do on RuntimeException. so in order to unify the shutdown and cleanup logic, I remove that catch block. so now the exception will be caught in the run method. it allows us to have the cleanup logic in one place instead of two

[mjsax]

because we don't stop the thread on StreamsException

We don't? This sounds off? A StreamsException is supposed to be treated as fatal... I find this a little bit confusing. Maybe I am missing something.

[Nikita-Shupletsov]

when we receive a StreamsException we shutdown the stream thread as well. it's just that a StreamsException is an expected exception, so we deal with it gracefully: we put the task in the list of failed tasks and then the TaskManager will deal with them in maybeThrowTaskExceptions. so we expect that something will throw a StreamsException and we have a graceful handling of it. We don't expect other kinds of exceptions/errors, so we treat them as a stop the world sort of event

[Nikita-Shupletsov]

future.get() blocks the StreamThread inside the rebalance callback with zero log output — the old code at least printed "The state updater wasn't able to remove task X in time." Suggest looping future.get(1, MINUTES) with a WARN per iteration and retrying forever: keeps the fix, restores observability. Same applies to shutdownStateUpdater(), where KafkaStreams.close(Duration) will now return false and leak the StreamThread instead of the previously-bounded 5-min-
per-future.

sounds good. will do

Integration test coverage regression. StateUpdaterFailureIntegrationTest was added by c48c50d (KAFKA-19831) specifically for "potential failures in Task#maybeCheckpoint". The PR deletes the flush() override that injected a ProcessorStateException during maybeCheckpointTasks and replaces it with a throw from onRestoreEnd, i.e. inside restoreTasks. Different handler — maybeCheckpointTasks catches StreamsException per task and continues; restoreTasks goes through handleStreamsException. The flush case should stay as a third Arguments rather than being replaced.

flush is no longer called because of the changes from https://issues.apache.org/jira/browse/KAFKA-19712. the original test kept passing because the failure was not induced anymore.
That's why I rewrote the test a bit, so we test the old case again faithfully.

StateUpdater.shutdown() semantics changed for restored tasks. failRemainingTasks() moves everything in restoredActiveTasks into the failed queue, and shouldShutdownStateUpdater now asserts that. Those tasks restored successfully; being "failed" means closeTaskDirty, discarding the checkpoint. It doesn't bite production today because shutdownStateUpdater() drains them via removeRestoredTask() before calling shutdown() — but it makes shutdown() unsafe to call on its own, which the unit test now enshrines as expected behavior.

so yes. it's another edge case my claude found: right now we don't and can't reuse a state updater. it lives as long as the thread it belongs to lives. but there was a unit test that was testing that we can shutdown and restart the same state updater which never happens in the real life. I decided to tightened it a bit to adhere to how we actually use it, to make it less errorprone. If we ever decide to reuse state updater or change the relationship between stream threads and state updater, there will at least be exceptions showing what we need to change instead of being in this gray zone.

[Nikita-Shupletsov]

Thank you @mjsax for reviewing the PR!

I answered and addressed yuor comments. please let me know if they make sense.

[mjsax]

Made another pass.

[mjsax]

Thanks for the fix. Merged to trunk and cherry-picked to 4.4 and 4.3 branches.