weiqingy opened a new issue, #645: URL: https://github.com/apache/flink-agents/issues/645
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description ## Summary `DurableExecutionManager.recordCheckpointSequenceNumbers()` unconditionally adds entries to `checkpointIdToSeqNums`, but `notifyCheckpointComplete()` only cleans them up when `actionStateStore != null`. When durable execution is disabled (`actionStateStore == null`), checkpoint entries accumulate indefinitely — a memory leak. ## Root Cause ```java // Always records — no null guard void recordCheckpointSequenceNumbers(long checkpointId, Map<Object, Long> seqNums) { checkpointIdToSeqNums.put(checkpointId, seqNums); } // Only cleans up when actionStateStore is non-null void notifyCheckpointComplete(long checkpointId) { if (actionStateStore != null) { ... checkpointIdToSeqNums.remove(checkpointId); // never reached when actionStateStore == null } } ``` ## Fix Add a `checkpointIdToSeqNums.remove(checkpointId)` call unconditionally in `notifyCheckpointComplete`, or guard `recordCheckpointSequenceNumbers` with an early return when `actionStateStore == null`. ## Notes - This bug existed before the `ActionExecutionOperator` refactoring (PR #546); it was surfaced during review of that PR by @wenjin272. - The fix and a test verifying the behaviour can be added in the same PR. ### How to reproduce This is a structural code bug — no specific input data is needed to trigger it. 1. Configure an agent job with durable execution **disabled** (i.e., do not set `ActionStateStore` — the default). 2. Enable Flink checkpointing (e.g., `env.enableCheckpointing(60_000)`). 3. Run the job and allow several checkpoints to complete. **Expected**: `checkpointIdToSeqNums` stays empty (no entries needed when durable execution is off). **Actual**: Each completed checkpoint adds an entry via `recordCheckpointSequenceNumbers()` that is never removed, because `notifyCheckpointComplete()` skips cleanup when `actionStateStore == null`. The map grows unbounded for the lifetime of the job. ### Version and environment - **Flink Agents**: 0.3-SNAPSHOT (identified in PR #546 review) - **Apache Flink**: 2.2.0 - **Java**: 11 - **Deployment mode**: any (bug is in operator logic, not deployment-specific) ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
