Thanks for the extensivee answer. I feel bad for writing a short answer only :(
Breaking existing clients is indeed a big concern, so I'm ok going with the approach you are proposing. I think we can open a vote for this FLIP soon. It has been open for some time, with no additional feedback. On Wed, Sep 9, 2026 at 12:55 AM Dale Lane <[email protected]> wrote: > > The only thing that I'm not fully decided on is the fact that the job > > submission would be blocking on the remote download. We could immdiately > > return the submission call and have the job in `INITIALIZING` state, if > > everything is ok, it would transition into CREATED, RUNNING, etc. > > If not, it would go into FAILED, and clients would have to fetch the > error > > from the job's exception endpoint. > > I do see benefits to that, including protecting against the sort of issue > described in FLINK-16866. > > My main motivation for the approach I pitched was that it only required > changes to core Flink, and kept current external behaviour. Existing > clients, such as the Flink Kubernetes Operator (FKO), continue working > as-is by just allowing users to include relevant new config options. (For > FKO that means users set the new config key on the session job and leave > jarURI unset, so it uploads its tiny placeholder jar and skips the download > it does today). > > It's true that the operator polls once it's in regular reconcile phase, > but the job submission phase is a long blocking call. FKO blocks on the > POST /jars/:jarid/run call, and will wait up to > kubernetes.operator.job.submission.timeout for it (10 minutes by default). > > All of this is why we could start supporting remote jar fetches from FKO > without requiring a coordinated change to benefit. > > I did include a nod to a potential second phase in "Rejected Alternatives": > > > Respond to submission requests immediately, and fetch and run jars > asynchronously > > This would be a change in behaviour and meaning to the endpoints as they > are today. (This is especially true for /plan where the response is the > plan generated from the fetched jar’s Job Graph). It would introduce the > need to have a way of learning the outcome of a fetch after the response > was sent, likely a new endpoint. This increases the size of the change that > would be needed. > > A new, typed jarUri field on JarRunRequestBody, with a companion POST > /jars/run endpoint that needs no {jarid} at all. > > This would result in a clearer self-documenting API, without requiring > clients to submit a jarid that will be ignored. It would be a more natural > fit for a value that names what job this is, not how Flink is configured. > > On balance, requiring a new endpoint will increase the effort needed to > adopt this, compared with supporting a configuration key available for all > clients already. Given that the goal is to remove work from orchestration > layers such as the Flink Kubernetes Operator, requiring a coordinated > update feels like it works against that goal. > > This could be addressed as a future piece of work, combined with the > point above, as the new endpoint could be designed to respond immediately, > without waiting for the fetch to complete. > > But I was being conservative, so let's consider this a little more. > > Looking at each of the handlers by endpoint, in terms of how hard it would > be to enable an asynchronous submission like you describe: > > **POST /jars/:jarid/plan** > As I put in the FLIP quote above, I don't think this would fit, as there > is nothing to poll a status on. > > **POST /jars/:jarid/run** > Today an HTTP-200 means the job was accepted with a graph. The web UI, > bin/flink run and the K8s Operator all rely on that. (FKO doesn't read the > jobid back out of the response - it supplies its own - but it does treat a > successful call as meaning the job is now there.) We'd be introducing > config options that change what that response means. > It'd complicate how we'd use JobStatus. JobStatus is described as > "possible states of a job once it has been accepted by the dispatcher" and > Dispatcher.requestJobStatus responds based on what is in the job-runner > registry or archived graph store. We'd be expanding INITIALIZING from > "received by the Dispatcher, waiting for the job manager to receive > leadership" to include a job without an execution graph yet. To do that, > we'd have to expand the Dispatcher to be able to distinguish between a job > that hasn't started yet from a job that's gone. > > **POST /jars/:jarid/run-application** > I think this one would work more easily. We get an applicationId up front > (either from the caller or generated by the handler) so we've got something > a submitter could poll on before main() is run. ApplicationState gives us a > state machine we could use as you describe, and a poll channel / error > channel. > > > I think this gives us a few options > > option 1) Update POST /jars/:jarid/run-application only, leave the other > two handlers as I proposed in the FLIP > I think this would be viable, but the K8s operator doesn't use this > endpoint today so it isn't useful for the FKO use case (without a > corresponding change to the FKO to use an alternative endpoint) > It'd also leave that endpoint meaning two things depending on whether a > jar URI was given - a fetch failure would stop being a 400 and become a > failed application instead. > That would be a substantial change to FKO, but there might be good reason > for FKO to start supporting the Application layer anyway (see the > discussion in > https://lists.apache.org/thread/2gs2xzzmh8371rnc2993o9bvjsm4n6bd - the > problem description "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'." is similar to > one of the challenges we'd face.) > > option 2) Create a new asynchronous fourth endpoint, with application as > the status resource > (what I described in the FLIP doc as "a companion POST /jars/run endpoint > that needs no {jarid}") > Also viable, and I prefer it to option 1, because it means we don't have > to worry about maintaining compatibility for any existing clients that use > the /jars/:jarid/run-application endpoint and we don't have to keep > submitting a placeholder jarid we don't use. > But, like option 1, it's also not useful for the FKO use case without > updating the Operator to use this new shape of API. > > option 3) Update POST /jars/:jarid/run (the endpoint that FKO uses) to > allow async submission > This wouldn't need a change to FKO, which is good. The trade off is that > we'd have to expand the Dispatcher to know about jobs that don't have an > execution graph yet, but if we could do that then the operator's existing > polling would find it in INITIALIZING and carry on as it does today. The > large number of Dispatcher touch points we'd need to update for this to > work put me off going down this path in the first place. > > > I'm obviously being shamelessly biased here, as I'm essentially ending up > back at the point where I started: go with a synchronous submission > approach to start with that is usable as-is by all clients including FKO, > and have a separate follow-on FLIP to add a more optimal asynchronous > route. (And the more recent discussion about Application layer support in > FKO suggests a good way that could be approached.) > > Most of the work I proposed in the FLIP would still be needed even if we > went with this approach to start with, so I don't think there is much > "wasted" work with going with this first phase. Fetching still happens on > the Job Manager either way, which is what I'd be building out. > > WDYT? > > Kind regards > > D > -- > dalelane.co.uk > > > On Tuesday, 8 September 2026 at 10:28, Robert Metzger <[email protected]> > wrote: > > > Overall, I'm +1 on the idea. > > > > The only thing that I'm not fully decided on is the fact that the job > > submission would be blocking on the remote download. We could immdiately > > return the submission call and have the job in `INITIALIZING` state, if > > everything is ok, it would transition into CREATED, RUNNING, etc. > > If not, it would go into FAILED, and clients would have to fetch the > error > > from the job's exception endpoint. > > > > It will make client implementations more difficult .. but the kubernetes > > operator is anyways polling already, so that should be fine. > > > > A long time ago, I worked on > > https://issues.apache.org/jira/browse/FLINK-16866. Where a > > low `web.timeout` caused issues with slow job submissions. Can you check > if > > your proposed implementation and defaults work well in the case where the > > remote artifact fetching for session clusters is stalling indefinetly? > Will > > we see the right error message on the client ("artifact fetching timed > > out") instead of ("job submission timed out")? > > > > > > On Mon, Aug 24, 2026 at 12:41 AM Dale Lane < > [email protected]> > > wrote: > > > > > I'd like to start a discussion on > > > FLIP-XXX : Remote Artifact Fetch for Session-Mode Job Submission > > > > > > > https://docs.google.com/document/d/1drZkXU875-j7lsyJaLp-VkVG_1PRietlfO2ETLeEWJc/edit?usp=sharing > > > > > > This FLIP proposes extending Flink's existing remote-artifact-fetch > > > capability to session-mode job submission, so that a job can be > submitted > > > by URI, with the Job Manager performing the fetch itself, the same way > an > > > Application Mode cluster already does at start-up. > > > > > > Looking forward to discussion - please let me know what you think! > > > > > > Kind regards > > > > > > Dale > > > -- > > > dalelane.co.uk > > > > > > > > >
