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
>

Reply via email to