Hi Hang, Thanks for the suggestion, that makes sense. Each coordinator adds at most one file per checkpoint, but I agree that sharing the threshold with task state is limiting, especially since data-inline-threshold cannot be set above 1 MB. I’ve added execution.checkpointing.coordinator-state-inline-threshold to the FLIP and the prototype, with no default. When it is not set, data-inline-threshold applies, as before. When it is set, it replaces data-inline-threshold for coordinator state only, and task state is unchanged. It has no maximum, since a larger value only keeps more coordinator state inline, which is what Flink does today. It only takes effect when execution.checkpointing.coordinator-state-in-file.enabled is true.
Best, Spoorthi Basu On Tue, Oct 13, 2026 at 3:33 AM Hang Ruan <[email protected]> wrote: > Hi Spoorthi, > > Thanks for the FLIP. Would it make sense to use a separate inline > threshold for coordinator state, for example: > > execution.checkpointing.coordinator-state-inline-threshold > > The existing execution.checkpointing.data-inline-threshold defaults to > 20 KB and also controls task state. Reusing it means that enabling this > feature may create many small coordinator state files, while increasing it > would also change task-state behavior. > > A separate threshold would allow users to balance JobManager memory and > checkpoint file overhead independently. WDYT? > > Best, > Hang > > On Mon, Oct 12, 2026 at 3:29 PM Yanquan Lv <[email protected]> wrote: > > > > Thanks for proposing this valuable optimization for large coordinator > > state. I’m concerned, however, about enabling > > execution.checkpointing.coordinator-state-in-file.enabled by default. > > Although V7 is written only when coordinator state is actually stored in > a > > file, a FLIP-27 source may begin producing V7 checkpoints automatically > > once its state exceeds the inline threshold, making those checkpoints and > > related HA entries unreadable by earlier Flink versions and reducing > > rollback safety immediately after an upgrade. The benefit is significant > > for large coordinator state but negligible when state remains inline. > Could > > we default this option to false initially, allowing users to opt in after > > the rollback observation period, and reconsider enabling it by default > once > > more production experience is available? > > > > > > Spoorthi Basu <[email protected]> 于2026年10月9日周五 10:01写道： > > > > > Hi all, > > > > > > Following up on this FLIP, which stores large coordinator state in its > own > > > file instead of inline in the checkpoint metadata. Any feedback on the > > > proposal would be much appreciated. > > > > > > Thank you, > > > Spoorthi Basu > > > > > > On Mon, Sep 28, 2026 at 6:07 PM Spoorthi Basu <[email protected]> > > > wrote: > > > > > > > Hi everyone, > > > > > > > > I'd like to start a discussion on FLIP-XXX: Store large operator > > > > coordinator state outside the checkpoint metadata [1]. > > > > > > > > Operator coordinator state, such as a source's split enumerator > state, is > > > > always stored inline in the checkpoint's _metadata file, whatever its > > > size. > > > > The JobManager also keeps a copy of it in memory for every retained > > > > checkpoint, and with high availability it is written a second time > into > > > HA > > > > storage on every checkpoint. As sources track more splits, this > state has > > > > grown large: an Iceberg user reports enumerator state past 1 GB [2]. > > > > > > > > The FLIP stores coordinator state above > > > > execution.checkpointing.data-inline-threshold in its own file, for > > > > coordinators that restore asynchronously (including every FLIP-27 > > > source). > > > > The coordinator reads the file itself when it is reset, off the > > > JobManager > > > > main thread, which is the condition FLINK-17674 named for relaxing > its > > > > restriction [3]. Only checkpoints that use this get a new metadata > > > version, > > > > and a new option turns it off for jobs that need to keep the current > > > > format. The FLIP does not change how much memory the enumerator > itself > > > > uses, or the 2 GB limit on a single coordinator's state. > > > > > > > > On a prototype [4], the _metadata of a job with 120 MB of enumerator > > > state > > > > goes from 120.8 MB to 1.4 KB, and a job with 30 MB of enumerator > state > > > and > > > > three retained checkpoints goes from 145 MiB to 58 MiB of used heap > in a > > > > local MiniCluster. > > > > > > > > Looking forward to your feedback. > > > > > > > > Best, > > > > Spoorthi Basu > > > > > > > > [1] > > > > > > > > https://docs.google.com/document/d/1AYWH6J-KsRGfrsMSy8Fi1kbREBiObPoZIb9JRnXhatU > > > > [2] https://issues.apache.org/jira/browse/FLINK-39754 > > > > [3] https://issues.apache.org/jira/browse/FLINK-17674 > > > > [4] > > > > > > > > https://github.com/spoorthibasu/flink/tree/flip-coordinator-state-prototype > > > > > > > >
