Hi Piotr, Thanks for the proposal. I also think this would be useful for large backlog catch-up cases.
I took a quick look at the current checkpoint and sink paths, and I think the questions Roman and Mate raised are the right ones to make explicit in the FLIP. For the checkpoint timeout point: today the timeout used by the coordinator is taken from the checkpoint coordinator configuration and the coordinator schedules a canceller for the created PendingCheckpoint. If the catch-up savepoint is created early and is expected to complete only when the bounded job reaches end-of-input, then the existing \`execution.checkpointing.timeout\` default looks easy to hit for long batch backfills. If the intended model is instead to trigger the savepoint only near the end of the batch execution, then this should be less of a problem. It would be helpful if the FLIP spelled out when the PendingCheckpoint is created and whether catch-up savepoints need a separate timeout or inherit the normal checkpoint timeout. For exactly-once sinks, I would avoid relying on an implicit "one transaction per sink subtask" rule unless that becomes part of the contract. In the current sink path, the writer prepares committables on checkpoint pre-barrier and also has an end-of-input path, while committer/global-committer stages may commit on \`notifyCheckpointComplete\` or when all committables are received in input-commit mode. For FLIP-605, I think the important part is to define which checkpoint or end-of-input event owns the final committables, and what happens to committables from a failed or restarted batch attempt before the streaming job restores from the savepoint. On max parallelism, I agree with Mate's concern. Keyed state restore depends on the max parallelism/key-group layout, and the current default max parallelism is derived from the operator parallelism when it is not configured. Since adaptive batch execution may infer different parallelism from the later streaming job, I think the safe contract should either require users to set the same \`pipeline.max-parallelism\` or operator max parallelism for both phases, or derive the batch job's max parallelism from the target streaming job specification. Otherwise this could become a subtle restore-time incompatibility. For checkpoint triggering, my reading is that the current coordinator sends RPC trigger requests only to the executions in the checkpoint plan's trigger set, and other tasks participate through the normal checkpoint propagation/lifecycle. If FLIP-605 needs non-source bounded tasks to snapshot state without the normal streaming barrier flow, it would be good to make that explicit. Finally, speculative execution probably needs an explicit rule. Since checkpoint acknowledgement and state are tied to execution attempts, the FLIP should either define which speculative attempt is allowed to contribute state for a subtask, or state that speculative execution is disabled for catch-up jobs that produce a savepoint. Best regards, Zhe Wang Mate Czagany <[email protected]> 于2026年8月15日周六 14:17写道： > Hi Piotr, > > Thank you for creating this FLIP, I think it would bring a lot of value for > the users where the streaming job has to reprocess a huge amount of > historical data. > > I have a few questions about the savepoints taken by the batch job, since > that seems to be where most of the effort will need to go with this FLIP > 1. When the streaming job restores from the savepoint taken by the batch > job, there will be requirements of the maximum parallelism of the > snapshotted job. The default max-parallelism calculated for streaming and > batch jobs will be different. Is there a plan to change that calculation? > Will the user be responsible to correctly set the max-parallelism for these > 2 jobs? > 2. Will there be any timeout for the `PendingCheckpoint` created? Currently > it's set to the value of `execution.checkpointing.timeout`. I guess with > the default value of 10 min, this would currently fail all batch jobs > running for longer than 10 minutes. > 3. Will the savepoint be able to handle attempts created by speculative > execution? Or should speculative execution be disabled if a savepoint needs > to be created? > > Best regards, > Mate > > On Fri, Aug 14, 2026 at 4:07 PM Roman Khachatryan <[email protected]> > wrote: > > > Hi Piotr, > > > > Thanks for the proposal! > > > > I have a couple of small questions: > > > > 1. Exactly-once sinks (let's say Kafka) > > Will every sink subtask have exactly 1 transaction to commit per > savepoint? > > Will there be any transactions to abort by the streaming job? Or are the > > transactions aborted on batch job restart? > > > > 2. Triggering checkpoint on a Task > > Will that be done via RPC for non-source tasks? Or is it a part of the > task > > lifecycle? > > > > > > Regards, > > Roman > > > > > > On Thu, Aug 13, 2026 at 8:12 AM Xuyang <[email protected]> wrote: > > > > > BTW, could you please share the branch for the PoC as well as some > > details > > > about the benchmark (e.g., setup and results) ? > > > > > > > > > > > > > > > > > > -- > > > > > > Best！ > > > Xuyang > > > > > > > > > > > > At 2026-08-12 14:22:22, "Xuyang" <[email protected]> wrote: > > > >Hi, Piotrek. > > > >Thanks a lot for putting this FLIP together. After modifying a > streaming > > > job, having to restart it statelessly often comes at a considerable > time > > > and resource cost for users. So big +1 from me. > > > >I've gone through the FLIP and it looks good to me overall. I just > have > > a > > > few questions I'd like to check with you: > > > >1. If the batch phase produces a savepoint in the standard format, my > > > understanding is that the streaming phase could restore from it > directly > > > via `state.savepoints.dir`. If that's true, would we still need to set > > > `execution.state-catch-up.mode: BATCH` in the streaming phase, or am I > > > missing something here? > > > >2. Regarding sources, I'm wondering whether it might have an interface > > to > > > indicate whether a source supports the bounded-to-unbounded transition. > > My > > > thinking is that existing bounded sources in current connectors don't > > > record their own state when they finish, so the subsequent streaming > > phase > > > would have nothing to pick up from. Curious whether you've already > > > considered how to handle this. > > > >3. I also have another rather rough idea around execution in the batch > > > phase, and I'd be very interested in your thoughts on it. For very > large > > > full datasets, a single failure midway can invalidate hours of earlier > > > batch computation. Even with fine-grained restarts in the batch phase > > down > > > the road, a single full-scale run would still need to retain the > blocking > > > shuffle partitions for the entire backlog until they're consumed,so > both > > TM > > > disk pressure and the recomputation cost would scale with the full > > dataset. > > > The idea would be to keep catch-up as a single job, but internally > split > > > the historical data into N shards(or N consecutive segments of data) > that > > > are consumed sequentially — producing a complete checkpoint as each > > > finishes, and producing the save point handed over to the streaming > phase > > > when the last one completes. This feels somewhat in the spirit of what > > > FLIP-309[1] describes, where checkpointing isn't disabled during > backlog > > > processing but simply happens at a lower frequency. That said, I may > well > > > be overlooking some complexity here, so please feel free to point out > > > anything I've missed. > > > >Thanks again for the great work! Looking forward to your thoughts. > > > > > > > > > > > >[1] > > > > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/255069517/FLIP-309+Support+using+larger+checkpointing+interval+when+source+is+processing+backlog > > > . > > > > > > > > > > > > > > > >-- > > > > > > > > Best！ > > > > Xuyang > > > > > > > > > > > > > > > >At 2026-08-07 21:53:16, "Piotr Nowojski" <[email protected]> > wrote: > > > >>Hi all! > > > >> > > > >>I would like to start a discussion on state catch-up using BATCH > > > execution > > > >>mode [1] > > > >> > > > >>FLIP-134 [2] introduced BATCH execution mode as a way to execute more > > > >>efficiently DataStream API jobs on bounded inputs. As even stated in > > the > > > >>documentation [3]: > > > >> > > > >>> One obvious outlier is when you want to use a bounded job to > > bootstrap > > > >>some job state that you > > > >>> then want to use in an unbounded job. For example, by running a > > bounded > > > >>job using > > > >>> STREAMING mode, taking a savepoint, and then restoring that > savepoint > > > on > > > >>an unbounded job. > > > >>> This is a very specific use case and one that might soon become > > > obsolete > > > >>when we allow > > > >>> producing a savepoint as additional output of a BATCH execution > job. > > > >> > > > >>producing a savepoint from BATCH execution job has always been > planned > > > as a > > > >>future extension. FLIP-605 [1] was created to implement this. > > > >> > > > >>The idea is that instead of waiting sometimes hours to process a long > > > >>backlog of records using STREAMING execution mode, users could submit > > > their > > > >>jobs in BATCH execution mode, create a savepoint at the end, and then > > > >>recover their jobs from that savepoint and continue running them in > > > >>STREAMING. Regardless if those jobs are using DataStream API or Flink > > > >>Streaming SQL/Table API. > > > >> > > > >>For more information please take a look into the FLIP document itself > > > [1]. > > > >> > > > >>Disclosure, in Confluent/IBM we have a working early access version > of > > > this > > > >>feature, hence we already have a pretty good understanding of what > > needs > > > to > > > >>be done to make it work and what are the expected results. > > > >> > > > >>Best, > > > >>Piotrek > > > >> > > > >>[1] > > > >> > > > > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/446071315/FLIP-605+State+catch-up+of+streaming+jobs+state+using+BATCH+execution+mode > > > >>[2] > > > >> > > > > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/158871522/FLIP-134+Batch+execution+for+the+DataStream+API > > > >>[3] > > > >> > > > > > > https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/execution_mode/#when-canshould-i-use-batch-execution-mode > > > > > >
