Hi Dale, Thanks, that is a thorough answer.
I checked the retry path in the code, since the new default relies on it. The idle event completes the request future with ConnectionIdleException. That extends ConnectionException, which extends IOException, and RestClusterClient's retry predicate ends with a catch-all on IOException, so an idle close is retried. rest.retry.delay is 3 seconds and rest.retry.max-attempts is 20, which matches the few seconds you saw. There is nothing equivalent on the server side. IdleStateHandler appears only in RestClient, and rest.idleness-timeout is read only by RestClientConfiguration. So a submitter that does not retry waits, rather than being disconnected by Flink. +1 from me, non binding. Best, Sai Krishna Sepuri On Mon, Oct 12, 2026 at 8:28 PM Dale Lane via dev <[email protected]> wrote: > > 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 > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
