Hi all, Thanks, Zhe, for confirming.
Following up on the wiki request above. Could a committer please create the FLIP page from the Google Doc and assign a number, or grant me write access to the Flink wiki? Best, Spoorthi Basu On Sat, Oct 17, 2026 at 10:45 PM Zhe Wang <[email protected]> wrote: > Hi Spoorthi, > > Thanks for addressing this and adding the blocked-read tests, including the > cases where the stream is still being opened. The cancellation behavior you > describe addresses my concern. I have no further questions on this point. > > Best regards, > Zhe Wang > > > Spoorthi Basu <[email protected]> 于2026年10月17日周六 03:48写道： > > > Hi Zhe, > > > > Thanks for raising this. It did not: a replaced reset, or one whose > > coordinator was closed, kept reading to the end, and its stream was only > > closed once the read finished. I’ve updated the FLIP and the prototype so > > that when a reset is replaced by a later one, or the coordinator is > closed, > > the read for that reset is stopped by closing its stream if it is still > in > > progress. Closing a stream can wait for a read in progress to return, so > it > > is done from a separate thread rather than the JobManager main thread. If > > the file is still being opened when the read is stopped, the stream is > > closed as soon as the open finishes. A stopped read does not fail the > job. > > I’ve also added a test with a stream whose reads block until it is > closed. > > It checks that the stream is closed and the reading thread finishes, for > a > > later reset and for closing the coordinator, each while the stream is > being > > read and while it is still being opened. > > > > Best, > > Spoorthi Basu > > > > On Fri, Oct 16, 2026 at 10:06 AM Zhe Wang <[email protected]> wrote: > > > > > 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 > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > > > > > > > > > >
