Hi Spoorthi, One remaining question about the lifecycle of asynchronous state reads: the FLIP explains that a superseded reset's result is discarded. Does superseding a reset, or closing the coordinator, also cancel the outstanding read and release its input stream? With large coordinator state and slow storage, repeated resets could otherwise leave multiple reads in flight even though only the latest result is needed. Could we cover this with a blocked-read test that checks resource cleanup, in addition to the existing test that verifies the later reset takes precedence?
Best, Zhe Wang Spoorthi Basu <[email protected]> 于2026年10月15日周四 20:27写道： > Hi all, > > Thanks everyone for the comments and suggestions. Since there are no > further concerns, I’d like to move this FLIP to the wiki and start a vote. > I’ve requested an Apache Confluence account (username: spoorthibasu). Could > a committer please approve it and grant write access to the Flink wiki, or > create the FLIP page and assign a number? > > Best, > Spoorthi Basu > > On Wed, Oct 14, 2026 at 8:09 AM Spoorthi Basu <[email protected]> > wrote: > > > Hi Sai Krishna, > > > > Thank you for the careful review and the +1. I appreciate you writing > down > > everything you checked. > > > > Best, > > Spoorthi Basu > > > > > > On Wed, Oct 14, 2026 at 4:45 AM Krishna Sai <[email protected]> wrote: > > > >> Hi Spoorthi, > >> > >> Thanks for the FLIP. I went through it along with the prototype branch, > >> looking at the parts I thought were riskiest, and found nothing to flag. > >> > >> What I checked, in case it saves someone a second pass: > >> > >> - Checkpoints#metadataSerializerFor picks the version on both write > >> paths that exist outside tests, PendingCheckpoint and > >> SavepointOutputFormat, so the State Processor API cannot write > >> version 6 metadata with a file handle in it. The older serializers > >> reject that on write too, not only on read. > >> - The write is in finalizeCheckpoint, so it is on the same thread that > >> already writes _metadata: the ioExecutor on the acknowledge path, the > >> checkpoint timer thread when the tasks have already acknowledged. > >> Never the main thread. > >> - File merging cannot pick up the file, since the merging location is > >> only built on the task side. > >> - The new threshold can do without a maximum because the stream it uses > >> passes a zero inline threshold, so unlike data-inline-threshold its > >> value never becomes the write buffer size. > >> - Savepoint disposal already deletes the file through > >> CheckpointMetadata#dispose, and StreamGraph is the only place that > >> builds JobCheckpointingSettings, so both options reach every job type. > >> > >> +1 from me, non binding. > >> > >> Best, > >> Sepuri Sai Krishna > >> > >> On Tue, Oct 13, 2026 at 1:03 PM 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 > >> > > > > > >> > > > > >> > > >
