Thanks Krishna for taking the time to look through and think about this. And I
think you're right that my defaults aren't sensible.
The detailed version of the FLIP has a table of the limits that apply to a
submission from the Kubernetes Operator, and how they relate to each other
("Submitting from the Flink Kubernetes Operator" at
https://cwiki.apache.org/confluence/spaces/FLINK/pages/451979327/FLIP-613+Remote+Artifact+Fetch+for+Session-Mode+Job+Submission+-+detailed#FLIP613:RemoteArtifactFetchforSessionModeJobSubmissiondetailed-SubmittingfromtheFlinkKubernetesOperator
)
The problem you've spotted is two of the defaults in the table are equal, so
there's a race that means the order they'll fire in isn't predictable.
I've done some experimenting with this, using FKO to run jobs that fetch from
an HTTP server that never responds.
With my defaults making user.artifacts.fetch-timeout equal to the Operator's
rest.idleness-timeout (5 minutes), I saw both of them timeout first. Sometimes
the fetch-timeout error reached the Operator, and sometimes the Operator closed
the idle connection a fraction of a second first. It's a race.
Making the two timeouts different made the behaviour predictable.
Making it smaller (I tried 4 minutes) - the error consistently reached the
Operator on the original connection
Making it larger (6 minutes) - the Operator's connection was closed as idle at
5 minutes, and its client retried the same request a few seconds later. The
extra minute allowed for on the Job Manager side means the retry was attached
to the submission still in progress (as described under "Retried submissions"
in the detailed FLIP), so there was no second fetch. The error naming the URI
and user.artifacts.fetch-timeout was consistently returned on the retry's
connection at 6 minutes.
In the FLIP, I'd suggested the former was better (even though I forgot that
when picking defaults!), as you did in your mail: making the fetch default
smaller.
Either way works, although having played with it, I think I prefer the second
approach, just because it doesn't restrict the longest fetch we allow below the
current 5 minutes.
I'll update the FLIP to change the default for user.artifacts.fetch-timeout to
6 minutes. I think this is the best fit for the Operator as it is today: it
gives a larger fetch budget than 5 minutes, and still leaves time within the
Operator's 10-minute job.submission.timeout for the job's main() to run.
I'll update the table in "Submitting from the Flink Kubernetes Operator" to
match that.
And I'll summarise this in the option's description: a submitter should either
keep the fetch timeout below its rest.idleness-timeout, or retry an idle
connection with the same request (asRestClusterClient does) so the retry is
attached to the submission in progress.
Thanks again for your time and for catching this.
Kind regards
D
--
dalelane.co.uk
On Monday, 12 October 2026 at 05:48, Krishna Sai <[email protected]> wrote:
> Hi Dale,
>
> On Robert's question about which error the client sees.
>
> Rejected Alternatives says "every existing client already waits on
> /run today; the fetch simply
> makes the wait longer". That wait has a ceiling: rest.idleness-timeout
> defaults to 5 minutes and
> RestClient applies it as a Netty IdleStateHandler on reader, writer
> and all-idle, so Flink's own
> client drops a connection idle that long. user.artifacts.fetch-timeout
> also defaults to 5 minutes.
> web.timeout is 10 minutes, so not the binding one.
>
> On stock defaults a fetch running to its deadline races the client's
> idle timeout. A fetch default
> below rest.idleness-timeout, or a note on the interaction, would make
> the artifact error the one
> that surfaces.
>
> Best,
> Sai Krishna Sepuri
>
> On Sun, Oct 11, 2026 at 11:33 PM Dale Lane via dev <[email protected]>
> wrote:
> >
> > Gentle nudge to try and revive the discussion on this one
> >
> > https://cwiki.apache.org/confluence/spaces/FLINK/pages/451974568/FLIP-613+Remote+Artifact+Fetch+for+Session-Mode+Job+Submission
> >
> > 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.
> >
> > Please take a look and share your thoughts
> >
> > Kind regards
> >
> > D
> > --
> > dalelane.co.uk
> >
> >
> > On Thursday, 10 September 2026 at 15:09, Robert Metzger
> > <[email protected]> wrote:
> >
> > > 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
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
>
