Hi Dennis Thanks for this - it's been useful.
I definitely agree with the big picture point (we should have a new story where we tackle application-centric reconciliation support) as I said above, and would love to help with that if I can. I also agree that a cleanup-complete signal in Flink core is a good idea. It would bring benefits such as stopping the Operator tearing down a JobManager in the middle of cleanup. In the meantime, I've followed through your steps to try and understand this. re: clean application results on recovery > ... on the recovery path, only > dirty application results are consulted, so a retained clean application > entry would not stop the job from re-running ... AIUI, if there is no recovered dirty result, Dispatcher#maybeSubmitApplicationInApplicationMode looks up the clean result (getCleanApplicationResultAsync) and only submits the application if there isn't one. https://github.com/apache/flink/blob/96d7fbe7deaa4ba723d1f08070730e2b4ebe1880/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherApplicationTest.java#L798-L802 I think this test is an example of this: - shutdown-on-application-finish=false - result written dirty then marked clean - asserts that main() is never called re: fixing the ordering I agree that fixing the ordering stops the upgrade issue on it's own, as long as delete-on-commit is left to true. Once cleanup removes the entry, the replacement cluster can start cleanly. I'm not sure that the ordering alone would address the scenario of a Job Manager restart after the application has finished (within the shutdown TTL). With delete-on-commit=true, nothing records the application finished so the restarted Job Manager runs a completed application again. Even if cleanup has time to complete, my understanding was that: * if delete-on-commit=true - both see nothing : upgrade works, but in the absence of any information saying the applicaiton finished, a restart re-runs main() * if delete-on-commit=false with a fixed path - both see the clean entry : upgrade is stuck, but the restart correctly recognises the application as complete * if delete-on-commit=false with a per-launch path - I think both should be handled correctly I think fixing the ordering would help the Operator from interrupting cleanup, but while clean entries are kept there is a benefit to the per-launch path, isn't there? Are you saying that the issue is tolerable in the meantime until an upstream core Flink improvement lands, or have I misunderstood the current issue? Kind regards D -- dalelane.co.uk On Monday, 14 September 2026 at 13:53, Dennis-Mircea Ciupitu <[email protected]> wrote: > Hi all, > > Thanks for bringing this up. This is an important topic and it deserves > care, so I took the time to reproduce the issue and to go through the whole > flow in detail, on both the operator and the Flink side, before forming an > opinion. > > I first want to lay out the operator cancellation flow as it is without any > of the fixes, then bring the old issue (FLINK-27569 [1]) and the current > one (FLINK-40467 [2]) into that picture. With both side by side it becomes > clear which kind of fix each of them actually needs. > > OPERATOR CANCELLATION FLOW PATH > > For presenting the operator cancelation path, I will use cancel with > savepoint, since this is the case highlighted in FLINK-27569 as well. > > Prerequisites: Flink 2.3, HA enabled, > execution.shutdown-on-application-finish=false (always set by the > operator), and both job-result-store.delete-on-commit and > application-result-store.delete-on-commit at their default of true. > > The concrete flow is: > > - Step 1: The operator triggers stop-with-savepoint and blocks until the > savepoint is written and the job is terminal. > - Step 2: On the Flink side, the job reaching a terminal state starts an > asynchronous chain that keeps running after the REST call has returned: > - 2.1. The job transitions to FINISHED, and its dirty entry, > <jobid>_DIRTY.json, is written to the job result store. > - 2.2. The application transitions to FINISHED and its dirty entry, > <appid>_DIRTY.json, is written to the application result store, while > the > job's HA data (job graph, blobs, checkpoint metadata) is cleaned up in > parallel. > - 2.3. Once both are done, the job entry is marked clean. > - 2.4. The application entry is removed from the HA store, the > application result is marked clean (deleted by default), and the > Dispatcher's application termination future completes. > - Step 3: The operator runs the redeployment flow: > - 3.1. The JobManager Deployment is scaled to zero and the operator > waits for the pods to disappear (bounded to at most one minute, > continuing > on timeout). This wait is for pod termination only, as it knows nothing > about Step 2. > - 3.2. The JobManager Deployment is deleted. > - 3.3. The HA ConfigMaps are deleted. > - 3.4. The CR status is patched to MISSING. > - 3.5. The operator creates the replacement cluster with the > savepoint path. > > Important note: Step 2 and Step 3 run concurrently. The operator only waits > for Step 1. Depending on storage latency and on how much there is to clean > up, Step 2 can be fully done before Step 3.1 (fast case), or still in the > middle when Step 3.1 sends SIGTERM to the JobManager (slow case). These two > timings are what the two issues are made of. > > The shape is the same for Native and Standalone. In Standalone, the > deletion is triggered directly at cancel time instead of at redeploy, so > the window is even shorter. > > Also note: Flink's own graceful shutdown > (shutdown-on-application-finish=true) waits for all of Step 2 before > exiting. The operator disables it on purpose, to keep observing the > terminal state through REST after the job has finished. > > OLD ISSUE (FLINK-27569) > > The old issue lived in the "Step 2 fast" timing plus an unexpected event: > Step 2 completed and deleted the job entry, then the JobManager crashed and > was restarted by its Deployment before 3.1 took place. Application mode > re-runs main() on every leader start with the same fixed job ID, found no > result store entry saying the job had already finished, and resubmitted the > job. The checkpoint store had already been cleaned in 2.2, so the job > started from an empty state and the savepoint just taken was ignored. > > This issue was in application mode only because application mode relies > entirely on the result store entry to decide whether the fixed job ID may > run again. Session mode recovers only the job graphs still present in the > HA store, and a finished job's graph is removed in 2.2, so nothing can > bring it back there. > > The fix at the time (FLINK-27573 [3]) set > job-result-store.delete-on-commit=false and pointed > job-result-store.storage-path at a random, per-launch directory under the > HA storage path. Both parts were required because retention kept the entry > alive for the whole lifetime of the JM deployment even after the > application mode job has finished, and the per-launch isolation prevented > that retained entry from blocking the next deployment, which at that time > reused both the HA cluster id and the fixed all-zero job ID. > > I still consider it the best available fix for that problem. What was > needed was an entry that outlives the job until the JobManager is torn > down. The operator cannot recreate the entry itself: it does not hold the > entry content, and it has no filesystem access to the HA storage. The Flink > 2.3 application layer does not help here either: on the recovery path, only > dirty application results are consulted, so a retained clean application > entry would not stop the job from re-running, and the job-level guard > remains the one that matters. The resource leak was the price, and it was > documented as such. > > Even if we want to re-think the fix for this issue now, that fix will fall > under Flink core and it will be too complicated and it doesn't worth the > risk. > > CURRENT ISSUE (FLINK-40467) > > The current issue lives in the "Step 2 slow" timing and needs no unexpected > event at all: Step 3.1 kills the JobManager after 2.3 has written > <appid>_DIRTY.json but before Step 2.4 has deleted it. The file survives on > the HA storage, and the HA metadata deletion in Step 3.3 does not touch it, > since it only removes ConfigMaps or ZooKeeper nodes. > > The replacement cluster then starts with the same HA cluster ID, hence the > same application-result-store path and, because the fixed Application ID is > derived from cluster.id, the same Application ID. The Dispatcher finds a > dirty result for its own Application ID, concludes the application already > terminated, resumes cleanup only and never resubmits the job. With > shutdown-on-application-finish=false the cluster stays up: healthy > JobManagers, working REST, the previous application shown as FINISHED, zero > jobs, and the operator reporting the job as missing. > > It is application mode only for the same structural reason: only > application mode derives a fixed Application ID from the HA cluster id and > re-runs main() from the bootstrap. In session mode the application id of a > job is the (rotated) job id by construction. > > COMPARING THE ISSUES > > The difference that matters is what the entry is supposed to do. In the old > issue the entry disappeared while it was still needed, and the only remedy > was to keep it for the whole deployment lifetime, which forces per-launch > isolation and the leak. In the current issue the entry is meant to be > deleted, by Flink, seconds after it is written, and the operator simply > does not let Flink finish. That is an ordering problem, and an ordering > problem is fixed by ordering, not by isolating the directory. > > Fixing it the same way as in 2022 (randomised > application-result-store.storage-path, PR #1199 [4]) works, but it fixes > the symptom and adopts a second permanent leak without the reason that > justified the first one. The operator also cannot delete those directories > later: the generated path is never stored in the status, and the operator > has no filesystem access to the HA storage. Still, this is the best > available fix that can be adopted in the operator only, but if we consider > fixing it at the root, Flink core, this will be pretty straightforward. > > What the operator actually needs is to know that Step 2 is complete before > executing Step 3.1. That signal does not exist today: the application > status turns FINISHED at the terminal transition, before Step 2.3 and Step > 2.4, and never changes afterwards. Adding it in Flink core is small and > purely additive. The Dispatcher already has a future that completes exactly > after Step 2.4 (the application termination future), and the application > details response already carries a timestamps map keyed by state name, so a > new key such as CLEANED can be added without any schema change on clients. > With it, the operator polls the existing /applications/:id endpoint after > stop-with-savepoint, with a timeout, and only then proceeds to Step 3.1. No > new endpoint, no leak, no manual cleanup, and gated to Flink 2.3+ in the > operator since the endpoint does not exist before. Making the cluster > shutdown REST call wait for cleanup instead would also be additive, but on > native Kubernetes that path makes Flink delete its own Deployment and HA > metadata, which changes the operator's lifecycle model rather than fixing > the race. > > The case raised in the reply (no high-availability.cluster-id configured, > every application ending up with the all-zero id) is the same identity > problem in another setting. The cleanup signal does not depend on identity > at all, so it covers that case as well. > > The one remaining scenario is the unexpected one: the JobManager dies in > the middle of Step 2 for reasons the operator did not cause, such as node > loss. Only there does a last resort similar to the job result store one > make sense, and it can be precise rather than unconditional. The operator > knows the expected job id of the new deployment, so if it observes a > healthy JobManager whose only application is terminal, with start and end > timestamps older than the Deployment itself, and no job with the expected > id, it has detected the wedge. The wedged cluster resumes the old cleanup > by itself and deletes the stale file when it completes, so the operator can > wait for the same CLEANED signal from it and then redeploy with the same > path. Only if the signal does not arrive within a timeout (cleanup retries > are unlimited by default) should it fall back to a one-off randomised path. > The redeploy itself reuses the existing resubmit path: fresh job id, > savepoint from the status, configuration from the current spec. The wedged > cluster consumed none of it, since main() never ran. > > APPLICATION ID IMPLICATIONS > > The application id is not the cause of the current issue and rotating it is > not the fix. The gap, teardown not waiting for cleanup, existed before 2.3. > The application result store made it visible because it is keyed by an > identifier that is stable across deployments and stored at a path that is > stable as well. The application id only plays a role in the last resort > detection described above. > > On the broader point of the thread I fully agree: the operator should > integrate with the application layer, since it should support every > submission and deployment feature Flink core supports. The right shape is > application-centric reconciliation for Flink 2.3+ deployments: an > ApplicationStatus in the CR status observed from the /applications > endpoints, cancel and upgrade decisions taken at application granularity, > and lifting the single-job cap that comes from > submit-failed-job-on-application-error=true so multi-job applications > become possible. That will be a separate story, and I would go for full > support there rather than observation only. > > WRAP UP > > To wrap up the things that I mentioned, I'd suggest going into the > following direction in order to resolve directly the issue at its root and > not to introduce intermediary steps in this approach that can bring > inconsistencies: > > - FLINK-40467 - the fix for this should fix the ordering. For this, we > need to add a cleanup-complete signal to the application details response > in Flink core, have the operator gate the teardown on it for 2.3+ > deployments, and keep a randomised path only as a detected, one-off > fallback for crashes during cleanup. > - Create a new story where we tackle application-centric reconciliation > support. > > Best, > Dennis > > [1] https://issues.apache.org/jira/browse/FLINK-27569 > [2] https://issues.apache.org/jira/browse/FLINK-40467 > [3] https://issues.apache.org/jira/browse/FLINK-27573 > [4] https://github.com/apache/flink-kubernetes-operator/pull/1199 > > On Mon, Sep 7, 2026 at 6:53 PM Dale Lane <[email protected]> > wrote: > > > Hi James > > > > Thanks for bringing this to the list - and for reviewing my PR on 1199 :) > > As I wrote the PR, I'm probably not the best person to answer whether the > > community should accept the approach, but I thought I'd add some more > > context. > > > > > > I've been working on FLIP-XXX "Running Flink jobs in MiniCluster using the > > Kubernetes Operator" [1] which uses FKO to run an application on an in-JVM > > MiniCluster rather than a distributed cluster. > > > > Building a PoC for my FlinkMiniCluster CR support, I found that nothing > > was setting high-availability.cluster-id because that path doesn't go > > through KubernetesClusterDescriptor (which is what sets it for a > > FlinkDeployment CR). FileSystemApplicationResultStore gets it's base path > > from high-availability.cluster-id, so every cluster's results ended up in > > the same directory. And ApplicationJobUtils.maybeFixIds derives the fixed > > ApplicationID from cluster.id, which it infers from > > high-availability.cluster-id. In the absence of a key, all of this got > > skipped, so in my first attempt every HA-enabled MiniCluster ran as > > application 00000000000000000000000000000000 > > > > The store's entry filename is applicationid.json, so both parts of the > > collision (the folder and the filename) are derived from the same > > high-availability.cluster-id option. > > > > It's a different topology and different route in to what prompted > > FLINK-40467, but the same collision. That's why I felt like I was > > addressing the cause, rather than a symptom, with my pull request. > > > > > > re: delete-on-commit > > I agree this needs doing. IMO a separate Jira issue would be good, as I > > think it's a different failure (JobManager restart inside the shutdown TTL, > > rather than a savepoint upgrade) which needs a different repro and test. > > > > re: resource leak > > Yeah, I agree that what I'm proposing in FLINK-40467 doubles down on the > > deal made in FLINK-27573, and maybe that's not a good thing. I used a > > random UUID because that's what I saw setRandomJobResultStorePath do, but > > in hindsight making directories anonymous makes the admin cleanup task > > harder as you can't derive the deployment from a directory name to know > > what is safe to remove. Perhaps the operator should record the path it > > generates in status - this would let us automate cleaning up the path for > > previous launches? > > > > > > re: modelling the application layer > > I think (a) and (b) are worth doing, irregardless of potential multi-job > > support benefits. > > > > The MiniCluster launcher PoC I created (described in [2]) was really > > really tiny, because the 2.3 Application layer did all the hard work: > > running the user's main() on the cluster, pinning the JobID, rejecting a > > resubmission of a job HA has already recovered, submitting a synthetic > > failed job when the application errors before it submits one, etc. I submit > > the same PackagedProgramApplication that > > ApplicationDispatcherGatewayServiceFactory builds for a distributed > > application cluster, but I got everything more or less for free. Better > > status reporting is a benefit we'd get from (b). The reason that > > "application fails before submitting a job" is visible today is > > submit-failed-job-on-application-error, which manufactures a job-shaped > > event so that a job-centric observer has something to see. That is a > > workaround for the issue you're describing - the missing application model. > > The operator can't afford to turn it off today, because then > > application-level failures go silent. > > > > More selfishly, if my FLIP is approved as currently written, then it'll > > introduce another CR that runs a Flink application and wants the same > > application-level status that (a) and (b) would make available. Obviously I > > am not trying to pre-empt how my FLIP goes, but just highlighting that I'd > > hope to be another consumer of what you're describing :) > > > > > > Happy to help with any of this if there is agreement > > > > [1] - https://lists.apache.org/thread/drz68pn4c7nd6tmojmoypvn39gtxdods > > [2] - > > https://docs.google.com/document/d/1dtGjPYcsBkx1vxHPs1QnDtPxeH_Acz_pl8gx4b1BLB4/edit?usp=sharing > > > > Kind regards > > > > D > > -- > > dalelane.co.uk > > > > > > > > On Tuesday, 1 September 2026 at 16:41, James Kan via dev < > > [email protected]> wrote: > > > > > Hi everyone, > > > > > > Flink 2.3 introduces the application as a first-class concept above jobs, > > > with its own > > > ID, state, and durable result store. The Kubernetes Operator is > > job-centric > > > and has no > > > representation of it. That produces one immediate bug and one longer-term > > > gap. I'd like > > > directional input on the first, and to open a discussion on the second. > > > > > > > > > 1. Deployments wedge on Flink 2.3 (FLINK-40467) > > > > > > On Flink 2.3 a deployment can come up healthy but never submit its job. > > The > > > JobManager > > > runs, REST responds, and the operator reports success -- nothing in the > > > logs says > > > otherwise. > > > > > > When an application terminates, Flink persists a terminal > > ApplicationResult > > > keyed by > > > Application ID, under a path derived from the HA cluster id. For the > > > operator both are the > > > CR name, which never changes across redeployments, so the next deployment > > > finds the old > > > record and Flink declines to re-run it. It is intermittent: Flink deletes > > > that record once > > > cleanup commits, and it only survives when the operator tears down the > > > JobManager while > > > cleanup is still running. > > > > > > There is an open PR for this already by contributors in the community: > > > https://github.com/apache/flink-kubernetes-operator/pull/1199 -- a > > unique > > > `application-result-store.storage-path` per deployment. Same shape as the > > > operator's > > > existing `setRandomJobResultStorePath` (FLINK-27569). We arrived at the > > > same fix > > > independently in our fork and have verified it on 2.3. > > > > > > The tradeoff is a deliberate resource leak: every deployment leaves a > > > directory behind that > > > nothing ever cleans up, and users are expected to prune them by hand. > > That > > > is the same deal > > > the job result store already makes, but this doubles it. > > > > > > *Question:* Is this an approach the community can accept? > > > > > > > > > 2. The operator has no model of the application layer > > > > > > Beyond this bug, the operator cannot observe or reason about applications > > > at all. There is > > > no Application ID in the CRD or status, and no use of the `/applications` > > > REST endpoints. > > > Consequences we have run into: > > > > > > - Application-level failures surface as "job not found", or not at all. > > > - Multi-job applications are unavailable. Under HA, 2.3 supports multiple > > > batch jobs per > > > application, but the operator sets > > > `submit-failed-job-on-application-error=true`, which > > > caps the job count at 1. > > > > > > If it is worth supporting, the operator would need to model the > > application > > > at some point, > > > and we are unsure how far that should go. Some directions, roughly in > > order > > > of how much > > > they change: > > > > > > (a) Observation only -- surface application state and its jobs in status, > > > emit events. > > > No behavioural change, and no CRD change beyond status. > > > (b) The operator acts on what it observes -- application state drives > > > lifecycle > > > decisions rather than a single job's state, so an application that is > > still > > > starting, one that failed before submitting a job, and one that was never > > > submitted > > > at all become distinguishable instead of all surfacing as "job not > > found". > > > (c) Fully application-centric -- lift the single-job assumption in the > > > reconciler and > > > the status model, which is what multi-batch actually requires. > > > > > > (a) and (b) seem useful on their own even if multi-job never happens, but > > > we may be missing > > > reasons not to. (c) is where we assume the real design questions are -- > > > what a per-job > > > status looks like, how snapshots and upgrades are addressed with more > > than > > > one job, and > > > what happens to `status.jobStatus`. > > > > > > So: is multi-job something the operator wants to support, and if so does > > > that ordering seem > > > reasonable? And is anyone already looking at this? We would love to hear > > > from the community. > > > > > > Thanks, > > > James > > > > > >
