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