ifndef-SleePy opened a new pull request, #9309: URL: https://github.com/apache/paimon/pull/9309
### Purpose This PR is a follow-up of [PIP-30 (#8220)]: it brings savepoint auto-tag to the coordinator-commit path for unaware-bucket append tables, so tagging behaves the same as on the existing committer-operator path. The goal is to match that behavior, not to add a new tagging model. ### Key design #### The coordinator cannot observe the savepoint, so the writer must report it to the coordinator The coordinator's checkpoint method does not tell it whether a checkpoint is a savepoint, so it cannot decide tagging on its own. The writer does see this at checkpoint time and already sends its committables to the coordinator, so it carries a savepoint flag on that report and the coordinator tags from it — reusing the existing writer → coordinator channel rather than adding a new one. #### The writer reports committables in `snapshotState` instead of `emitCommittables` The writer only learns whether a checkpoint is a savepoint in `snapshotState`, which runs after `emitCommittables`, so the report moves there. This is safe: the report is an RPC message to the coordinator and does not depend on the checkpoint barrier. #### The coordinator rebuilds the pending tags after failover instead of checkpointing them The coordinator's checkpoint is taken before the writer sends the savepoint flag, so it cannot store the pending tags in its own state. Instead it rebuilds them from the flags the writers replay on restore — the same replay they already need for the commit to be correct, so there is no extra state to persist and no second source of truth that can drift. #### When the tag is created (async vs sync savepoint) An async savepoint does not get its own completion notification ([FLIP-193](https://cwiki.apache.org/confluence/display/FLINK/FLIP-193:+Snapshots+ownership)), so its tag is created by a later checkpoint. A sync savepoint (`stop-with-savepoint`) does get its own completion, but the coordinator path cannot tag it yet, for two reasons: it lacks the end-input handling a terminating savepoint needs, and because commits run on the coordinator's asynchronous executor, the shutdown that follows stop-with-savepoint can outrun that final completion and drop it. This is a known gap left to the [follow-up end-input PR (#8671)](https://github.com/apache/paimon/pull/8671), with an IT added but disabled. #### An empty async savepoint produces no tag, matching the operator path An async savepoint that carries no data produces no tag; this is a pre-existing limitation of the operator path that the coordinator path inherits, not a new gap. A tag needs a committed snapshot whose id equals the savepoint's id. For an empty savepoint that snapshot is never produced: the async savepoint's id never gets a completion notification of its own (above), and `commit.force-create-snapshot` only creates an empty snapshot for an id that is notified (no committable for the savepoint ever enters `committablesPerCheckpoint`). So while the ordinary empty checkpoints around it each get a snapshot, the savepoint's id is skipped, and the later catch-up finds nothing to tag — `findSnapshotsForIdentifiers` cannot find a snapshot for the savepoint's id. ### Consistency A savepoint tag ends up present only if the savepoint's snapshot was committed and the savepoint was not aborted. This holds across restart and failover: - **Async savepoint succeeds.** A later checkpoint creates its tag. - **Restore when the tag already exists.** Recovery re-runs tagging, but creating a tag that already exists is ignored, so it is idempotent. - **Restore when the tag was never created.** The writer replays the flag and the first commit after restore creates the tag. - **Savepoint aborted (async), no failover.** A failed async savepoint does fire `notifyCheckpointAborted` for its id, unlike a successful one, which fires no notification at all. The coordinator uses that as its cue to drop the pending flag and remove any tag a later checkpoint had already created. - **Region failover with an in-flight savepoint.** The coordinator only commits on checkpoint completion, and the interrupted savepoint's checkpoint never completes, the coordinator clears the aborted savepoints, so it commits nothing and stays untagged. - **Global failover with an in-flight savepoint.** The writer also clears the aborted savepoint's flag, so restoring from a later checkpoint does not recreate a tag for a savepoint that no longer exists. ### Tests - `AppendTableSavepointTagFailoverITCase` — failover and restore coverage: re-tag on restore when the tag already exists and when it never existed (both paths), plus coordinator-only region-failover and global-failover cases. - `AppendTableSavepointTagITCase` — async savepoint tagging on both paths, plus a disabled `stop-with-savepoint` case kept for the end-input follow-up PR. - Unit tests: `SavepointTaggerTest`, `CommittingWriteOperatorCoordinatorTest`, `CoordinatorCommittingRowDataStoreWriteOperatorTest`, `CheckpointCommittablesSerializerTest`, `FlinkSinkTest`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
