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
> >
>
