Hey! Thanks for the analysis of the issue. I agree that in the long run we should try to find a good solution that ideally supports the application concept properly (multi job deployments) and also doesn't leak storage unnecessarily.
I feel however that this would require a larger change to the controller flow and would require a FLIP. In the meantime since currently the operator doesn't rely on the application concept during reconciliation/cleanup, we should follow the currently established leaky solution with the random paths as it would properly unblock 2.3 single job deployments the same way as with previous flink versions. Cheers Gyula On Mon, Sep 14, 2026 at 5:47 PM Dale Lane <[email protected]> wrote: > > 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 > > > > > > > > >
