Hello all, I’ve discovered a gap in the stop-with-savepoint lifecycle, which could potentially cause data duplication or incorrect job state when using with Flink Kubernetes Operator. I would like to receive some early feedbacks before writing a FLIP. Because the fix I have prototyped adds public config keys.
## Problem After a stop-with-savepoint, the savepoint path exists only in the JobManager’s memory (savepoints are deliberately not added to the CompletedCheckpointStore). An external controller such as the Kubernetes operator polls and reads that path via JM’s REST API. Which leaves this window open for a race condition: 1. Stop-with-savepoint completes, the savepoint is persisted to filesystem, job state FINISHED, Dispatcher.jobReachedTerminalState run and marks the job as dirty. 2. JM crashes before the operator can poll 3. The restarted JM finds the dirty JobResultStore entry and runs cleanup. The job’s HA data is removed from HA store. 4. The savepoint is still on filesystem, but nothing records its path. The operator, during the next poll, now finds that /jobs/<id>/checkpoints return 404 5. The operator, depending on the setting, may now try to recover stateless or from an older checkpoint. This has caused data duplication and incorrect job state for us in production. To recover, a human must manually locate the savepoint directory. The HistoryServer (if enabled) does record the path, but the JobManager REST API never talks to it, so external clients that only talk to the JM cannot recover. ## Proposal Two new config keys: 1. jobmanager.completed-jobs.persist-dir (string, unset = disabled): a durable directory for job manager to persist the ExecutionGraphInfo of a globally terminated job, **before** registering it in JobResultStore. 2. jobmanager.completed-jobs.persist-retention (duration, default 72h or some sensible value): JobManager cleans up entries older than this Dispatcher.onStart repopulates the ExecutionGraphInfoStore from that directory, so even after a JM restart, REST API can serve the terminated jobs’ state (also show up in web UI). These states include checkpoint history and the final savepoint path. I have made a prototype, and empirically verified that it works. Job states are now persisted across JM restarts for FINISHED jobs. This works for both REST API and the UI. See commit link here: https://github.com/sqd/flink/commit/f7213593885aff4558218368ddaaf30801655a11 (disclaimer: this code is AI assisted, if this proposal moves forward I’ll rewrite and clean up). This is against Flink 2.0.1 because it’s the easiest to test for me, but I have verified that relevant logics remain identical in 2.3.0. ## Questions: 1. New option, or derive the directory from high-availability.storageDir? Arguably this is data used for HA 2. Alternatively, we could wire the REST API to fall back to the history server? (jobmanager.archive.fs.dir) 3. Granularity: store the whole ExecutionGraphInfo, or just the final savepoint path? 4. Default off or enabled together with HA? Best, Han
