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
