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