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