From 71513de0248a75d0dca0456502ddc8d53b76b6f0 Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Tue, 12 May 2026 21:41:58 -0700 Subject: [PATCH] [runtime] Lock null-store symmetry invariant in DurableExecutionManager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #645 originally described a memory leak in checkpointIdToSeqNums when actionStateStore == null: an unconditional put on the snapshot side paired with a guarded remove on the cleanup side, causing the map to grow unboundedly when durable execution was disabled. That asymmetry was structurally fixed during the DurableExecutionManager extraction in #546, which added a symmetric actionStateStore == null early-return at the top of snapshotLastCompletedSequenceNumbers. The symmetry is now held by two independent conditionals with no test or javadoc tying them together — a future refactor that drops either guard would silently reintroduce the leak. This commit locks the invariant in place: - Add a @VisibleForTesting accessor for checkpointIdToSeqNums mirroring the existing getActionStateStore() precedent. - Add a regression test that, with actionStateStore == null, runs two snapshot + notifyCheckpointComplete cycles and asserts the map stays empty. verifyNoInteractions on the mock KeyedStateBackend additionally proves the snapshot-side early-return fires before any backend access. - Strengthen javadoc on both methods to call out the symmetric guard invariant and cross-link to issue #645. No production behavior change. --- .../operator/DurableExecutionManager.java | 12 +++++++++ .../operator/DurableExecutionManagerTest.java | 25 +++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/DurableExecutionManager.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/DurableExecutionManager.java index 03bfe2643..85c5df23a 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/DurableExecutionManager.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/DurableExecutionManager.java @@ -333,6 +333,12 @@ void maybePruneState(Object key, long sequenceNum) throws Exception { * via {@link #snapshotLastCompletedSequenceNumbers}. After pruning, the entry for that * checkpoint is removed. No-op when durable execution is disabled. * + *

Invariant: the {@code checkpointIdToSeqNums.remove} below and the {@code put} in + * {@link #snapshotLastCompletedSequenceNumbers} MUST share the same {@code actionStateStore != + * null} guard. Dropping the guard on either side breaks the symmetry and reintroduces the + * unbounded-map leak tracked by + * issue #645. + * * @param checkpointId the id of the completed checkpoint. */ void notifyCheckpointComplete(long checkpointId) { @@ -365,6 +371,12 @@ void snapshotRecoveryMarker() throws Exception { * strictly up to the sequence number that was committed by that checkpoint. No-op when durable * execution is disabled. * + *

Invariant: the {@code checkpointIdToSeqNums.put} below and the {@code remove} in + * {@link #notifyCheckpointComplete(long)} MUST share the same {@code actionStateStore != null} + * guard. Dropping the guard on either side breaks the symmetry and reintroduces the + * unbounded-map leak tracked by + * issue #645. + * * @param keyedStateBackend the keyed state backend to scan. * @param checkpointId the id of the checkpoint being snapshotted. */ diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/DurableExecutionManagerTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/DurableExecutionManagerTest.java index 6c737d47b..f5fc2a172 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/DurableExecutionManagerTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/DurableExecutionManagerTest.java @@ -43,6 +43,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; /** Contract tests for {@link DurableExecutionManager}. */ @@ -69,6 +70,30 @@ void noStoreModeMakesAllMaybeOperationsNoOp() throws Exception { dem.close(); } + @Test + @SuppressWarnings("unchecked") + void noStoreModeSnapshotAndNotifyKeepCheckpointMapEmpty() throws Exception { + DurableExecutionManager dem = new DurableExecutionManager(null); + KeyedStateBackend backend = mock(KeyedStateBackend.class); + + // Cycle 1: snapshot + notify with null store. The snapshot-side guard must short-circuit + // before any backend access, and the cleanup-side guard must leave the map untouched. + dem.snapshotLastCompletedSequenceNumbers(backend, 1L); + assertThat(dem.getCheckpointIdToSeqNums()).isEmpty(); + verifyNoInteractions(backend); + dem.notifyCheckpointComplete(1L); + assertThat(dem.getCheckpointIdToSeqNums()).isEmpty(); + + // Cycle 2: confirm the invariant holds across multiple checkpoints. + dem.snapshotLastCompletedSequenceNumbers(backend, 2L); + assertThat(dem.getCheckpointIdToSeqNums()).isEmpty(); + verifyNoInteractions(backend); + dem.notifyCheckpointComplete(2L); + assertThat(dem.getCheckpointIdToSeqNums()).isEmpty(); + + dem.close(); + } + @Test void withInjectedStorePersistsTaskResult() throws Exception { InMemoryActionStateStore store = new InMemoryActionStateStore(false);