You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by Spoorthi Basu <sp...@gmail.com> on 2026/10/09 01:58:45 UTC
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
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 <sp...@gmail.com>
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
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
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 <ja...@gmail.com> 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 <sp...@gmail.com> 于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 <ja...@gmail.com> 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 <sp...@gmail.com> 于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 <
> spoorthibasu@gmail.com>
> > > > 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 <sa...@gmail.com>
> > > 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 <
> ruanhang1993@apache.org>
> > > > >> 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 <ku...@apache.org>
> > > 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 <sp...@gmail.com> 于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 <
> > > > >> spoorthibasu@gmail.com>
> > > > >> > > > 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
> > > > >> > > > >
> > > > >> > > >
> > > > >>
> > > > >
> > > >
> > >
> >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Zhe Wang <ja...@gmail.com>.
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 <sp...@gmail.com> 于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 <ja...@gmail.com> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > > 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 <sa...@gmail.com>
> > 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 <ru...@apache.org>
> > > >> 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 <ku...@apache.org>
> > 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 <sp...@gmail.com> 于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 <
> > > >> spoorthibasu@gmail.com>
> > > >> > > > 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
> > > >> > > > >
> > > >> > > >
> > > >>
> > > >
> > >
> >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
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 <ja...@gmail.com> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > 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 <sa...@gmail.com>
> 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 <ru...@apache.org>
> > >> 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 <ku...@apache.org>
> 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 <sp...@gmail.com> 于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 <
> > >> spoorthibasu@gmail.com>
> > >> > > > 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
> > >> > > > >
> > >> > > >
> > >>
> > >
> >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Zhe Wang <ja...@gmail.com>.
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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> 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 <sa...@gmail.com> 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 <ru...@apache.org>
> >> 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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <
> >> spoorthibasu@gmail.com>
> >> > > > 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
> >> > > > >
> >> > > >
> >>
> >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
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 <sp...@gmail.com>
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 <sa...@gmail.com> 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 <ru...@apache.org>
>> 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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <
>> spoorthibasu@gmail.com>
>> > > > 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
>> > > > >
>> > > >
>>
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
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 <sa...@gmail.com> 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 <ru...@apache.org> 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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <
> spoorthibasu@gmail.com>
> > > > 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
> > > > >
> > > >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Krishna Sai <sa...@gmail.com>.
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 <ru...@apache.org> 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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > > 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
> > > >
> > >
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
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 <ru...@apache.org> 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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > > 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
> > > >
> > >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Hang Ruan <ru...@apache.org>.
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 <ku...@apache.org> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > 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
> > >
> >
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Spoorthi Basu <sp...@gmail.com>.
Hi Yanquan,
Thanks for the feedback, that’s a fair point. I’ve updated the FLIP to
default execution.checkpointing.coordinator-state-in-file.enabled to false.
With it disabled, every checkpoint and savepoint stays in the format Flink
writes without this FLIP, and HA entries stay readable as they are today,
so the FLIP does not affect rolling back unless the option is enabled.
Users with large coordinator state can opt in after the rollback
observation period, and the FLIP notes that enabling it by default can be
reconsidered once there is more production experience.
Best,
Spoorthi Basu
On Mon, Oct 12, 2026 at 3:29 AM Yanquan Lv <ku...@apache.org> 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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> > 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
> > >
> >
>
Re: [DISCUSS] FLIP-XXX: Store large operator coordinator state outside the checkpoint metadata
Posted by Yanquan Lv <ku...@apache.org>.
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 <sp...@gmail.com> 于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 <sp...@gmail.com>
> 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
> >
>