KAFKA-19831: Improved error handling in DefaultStateUpdater. #20767

Nikita-Shupletsov · 2025-10-24T23:11:55Z

Improved error handling in DefaultStateUpdater to take potential failures in Task#maybeCheckpoint into account.
Improved TaskManager#shutdownStateUpdater to not hang indefinitely if the State Updater thread is dead.

Nikita-Shupletsov · 2025-10-24T23:14:47Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskAndAction.java

        return new TaskAndAction(null, taskId, Action.REMOVE, future);
    }

-    public static TaskAndAction createRemoveTask(final TaskId taskId) {


the non future remove task isn't used anywhere anymore, so I deleted the code that handled it.
also it made the proposed solution a bit tidier

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

mjsax · 2025-11-06T00:08:31Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

-            measureCheckpointLatency(() -> task.maybeCheckpoint(true));
+                // we need to enforce a checkpoint that removes the corrupted partitions
+                measureCheckpointLatency(() -> task.maybeCheckpoint(true));
+            } catch (final StreamsException e) {


In general, we treat StreamsException as fatal, and should never swallow it. What is the reason for this change?

Also: maybe rename e -> swallow to make clear that we swallow intentionally, and not by mistake.

because it's for a corrupted task. corrupted tasks already have an exception, so to not overwrite it, I just log it here.
renaming e to swallow is a good idea, will do

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

mjsax · 2025-11-06T00:29:49Z

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

+
+                    @Override
+                    public void flush() {
+                        if (numberOfStoreInits.get() == NUM_PARTITIONS * 1.5) {


This could be brittle? Should we just hardcode 9? If NUM_PARTITIONS is changed it's not guaranteed we end up with "integer". Also, the == would be evaluated on double type right now -- don't think we would want this.

mjsax · 2025-11-06T00:31:12Z

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

+
+        streams.removeStreamThread();
+
+        TestUtils.waitForCondition(() -> numberOfStoreInits.get() == NUM_PARTITIONS * 1.5, "Streams never reinitialized the store enough times");


Similar comment.

mjsax · 2025-11-06T00:31:38Z

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

+                    public void flush() {
+                        if (numberOfStoreInits.get() == NUM_PARTITIONS * 1.5) {
+                            try {
+                                TestUtils.waitForCondition(() -> currentState.get() == KafkaStreams.State.PENDING_SHUTDOWN, "Streams never reached PENDING_SHUTDOWN state");


Not sure if I understand this condition. Can you elaborate?

Also: if we fail here, how do we ensure that the test fails? This would be executed on the background StreamsThread not the actual "main" thread running the test.

to reproduce the issue, we need to fail during a shutdown.

so what the test does:

it creates a KS app with 2 threads.

then kills one thread which causes a rebalance

then the rebalance hangs here, because we can't flush a task.

we call streams close

this call gets unblocked and throws an exception

that's where we reproduce the issue

in other words: the problem I am trying to fix here reproduces if: we are in rebalance, we have some unhandled tasks in the task updater which can throw an exception, which can kill the thread, and we are trying to close the app.
in this case the stream thread will not do its thing, which in turn will not call TaskManager, which will not call stateUpdater.drainExceptionsAndFailedTasks, which will not remove all failed tasks from the lists. which in turn will cause stateUpdater.tasks to return the failed tasks. and TaskManager will try to delete them.
but because the thread is dead, the delete will never succeed, as the dead thread will never complete the future

I will add a comment to the test to explain what exactly we are trying to achieve here

I understand that we do want to throw here to trigger an error on the state-updater thread. But it's not clear to me, why we would need to block/wait here until we go to "pending shutdown"?

what needs to happen to reproduce the issue:
the stream thread is already down(so, we want this condition: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L933 to be false), we also want to have a task in the state updater. and this task needs to throw an exception.

maybe there is an easier way to achieve these conditions, but that's what I was able to come up with

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

mjsax · 2025-11-06T00:38:16Z

...s/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java

+    }
+
+    @Test
+    public void correctlyHandleFlushErrorsDuringRebalance() throws InterruptedException {


This test is somewhat complex. Might be good to add a longer comment to explain what it does.

If I read it correctly, we start up two thread, and each thread get 3 tasks each (and we init 6 stores). We stop on thread, re-trigger the init of 3 stores, and let it crash when we hit "store init no.9".

mjsax · 2025-11-06T00:42:41Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

        final StateUpdater.RemovedTaskResult removedTaskResult;
        try {
-            removedTaskResult = future.get();
+            removedTaskResult = future.get(1, TimeUnit.MINUTES);


Do I understand correctly, that this is the core fix, to avoid blocking forever?

(Plus the additional cleanup code add to shutdownStateUpdater in case we really hit the timeout here?)

the core fix is the error handling. my idea is to never let the state updated thread die if we get a StreamsException. if we don't die and handle them correctly, we will not need to worry about hanging here.

I added a timeout here just an extra layer of protection. that's why I added BUG_ERROR_MESSAGE to the log message, because I don't think we should expect it ever happen under normal circumstances

Co-authored-by: Matthias J. Sax <[email protected]>

Conflicts: streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

…ng tested. Small refactoring.

KAFKA-19831: Improved error handling in DefaultStateUpdater.

d549788

github-actions bot added triage PRs from the community streams labels Oct 24, 2025

Nikita-Shupletsov commented Oct 24, 2025

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java Show resolved Hide resolved

github-actions bot removed the triage PRs from the community label Oct 25, 2025