[Bug][runtime] Fix memory leak in DurableExecutionManager.checkpointIdToSeqNums

[weiqingy]

Search before asking

• I searched in the 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

// 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 [runtime] Refactor ActionExecutionOperator into focused manager classes #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 [runtime] Refactor ActionExecutionOperator into focused manager classes #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?

• I'm willing to submit a PR!