KAFKA-17402: DefaultStateUpdated should transite task atomically (#18607

) Reviewers: Bruno Cadonna <[email protected]>, Lucas Brutschy <[email protected]>
apache · Jan 21, 2025 · 8609ce9 · 8609ce9
1 parent 2d541ed
commit 8609ce9
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java
@@ -674,7 +674,6 @@ private void maybeCompleteRestoration(final StreamTask task,
                 measureCheckpointLatency(() -> task.maybeCheckpoint(true));
                 changelogReader.unregister(changelogPartitions);
                 addToRestoredTasks(task);
-                updatingTasks.remove(task.id());
                 log.info("Stateful active task " + task.id() + " completed restoration");
                 transitToUpdateStandbysIfOnlyStandbysLeft();
             }
@@ -690,6 +689,7 @@ private void addToRestoredTasks(final StreamTask task) {
             restoredActiveTasksLock.lock();
             try {
                 restoredActiveTasks.add(task);
+                updatingTasks.remove(task.id());
                 log.debug("Active task " + task.id() + " was added to the restored tasks");
                 restoredActiveTasksCondition.signalAll();
             } finally {