You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "ifndef-SleePy (via GitHub)" <gi...@apache.org> on 2026/08/19 19:41:57 UTC
[PR] [flink] Support savepoint auto-tag on coordinator commit path [paimon]
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support savepoint auto-tag on coordinator commit path [paimon]
Posted by "ifndef-SleePy (via GitHub)" <gi...@apache.org>.
ifndef-SleePy commented on code in PR #9309:
URL: https://github.com/apache/paimon/pull/9309#discussion_r3827977344
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.sink.FlinkSinkBuilder;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.util.AbstractTestBase;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end savepoint auto-tag tests for unaware-bucket append tables, parameterized over the two
+ * commit paths (coordinator-commit and the classic global committer). Both must create the same
+ * {@code savepoint-<checkpointId>} tag for a triggered savepoint.
+ *
+ * <p>Uses a source that emits continuously so the async savepoint deterministically lands on a
+ * data-carrying checkpoint; the empty-savepoint boundary (a separate, shared limitation) is
+ * intentionally avoided here.
+ */
+public class AppendTableSavepointTagITCase extends AbstractTestBase {
+
+ // The savepoint tag only materializes once a checkpoint *after* the savepoint completes and
+ // cumulatively commits the savepoint's snapshot (same catch-up as the classic path). Give the
+ // poll generous headroom so a transient checkpoint stall under load cannot trip the assertion.
+ private static final long WAIT_TIMEOUT_MILLIS = 120_000L;
+
+ @ParameterizedTest(name = "coordinatorCommit = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testSavepointCreatesTag(boolean coordinatorCommit) throws Exception {
+ String tableName = coordinatorCommit ? "T_COORD" : "T_CLASSIC";
+ FileStoreTable table = createTable(tableName, coordinatorCommit);
+
+ JobClient client = runSink(table);
+ try {
+ // Wait until a data-carrying snapshot exists so the async savepoint that follows
+ // deterministically lands on a checkpoint that carries data.
+ waitUntilSnapshotWithData(table);
+
+ client.triggerSavepoint(
+ getTempDirPath("savepoint_" + tableName), SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS);
+
+ // Poll until exactly one savepoint-prefixed tag appears, then assert it is consistent
+ // with the snapshot it points at.
+ Map<Snapshot, List<String>> savepointTags = waitUntilSavepointTagCreated(table);
+ assertThat(savepointTags).hasSize(1);
+ Map.Entry<Snapshot, List<String>> snapshotWithTags =
+ savepointTags.entrySet().iterator().next();
+ Snapshot tagged = snapshotWithTags.getKey();
+ assertThat(snapshotWithTags.getValue())
+ .containsExactly(SavepointTagUtils.tagNameOf(tagged.commitIdentifier()));
+ assertThat(table.snapshotManager().snapshotExists(tagged.id())).isTrue();
+ } finally {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * A sync savepoint (stop-with-savepoint) receives its own {@code notifyCheckpointComplete},
+ * unlike an async savepoint, so the tag is created for the savepoint's own snapshot rather than
+ * caught up by a later checkpoint. Both commit paths must still produce the same tag.
+ *
+ * <p>Disabled for now: on the coordinator-commit path this is racy. The coordinator creates the
+ * tag asynchronously on its single-thread commit executor (notifyCheckpointComplete ->
+ * tagUpTo), but stop-with-savepoint terminates the job right after the savepoint, and the
+ * coordinator's {@code close()} calls {@code commitExecutor.shutdownNow()}, which can drop the
+ * not-yet-run tag task so the tag is silently lost. The classic operator path is unaffected
+ * because it tags synchronously. Re-enable once the coordinator drains pending commit/tag work
+ * on end-of-input shutdown (the follow-up PR that adds proper end-input handling to the
+ * coordinator).
+ */
+ // TODO: enable once the coordinator supports end-input handling (drains pending tag work).
+ @Disabled(
Review Comment:
Yes, this is a real issue.
I looked into using a drain as a lightweight fix for stop-with-savepoint when auto-tagging is enabled with coordinator commit, before end-input support is available. Draining the executor in `CommittingWriteOperatorCoordinator.close()` should work in the normal case.
**However, if processing the final savepoint completion fails to create the tag, it cannot provide the same guarantee as the operator path.**
1. For coordinator commit, `SchedulerBase.closeAsync` disposes operator coordinators through `IOUtils.closeQuietly`, which ignores exceptions from `close()`. A failed drain could therefore leave the savepoint successful and the job finished, while the tag was never created.
2. For operator commit, `StreamTask.afterInvoke` waits for `finalCheckpointCompleted` before finishing. If the final `notifyCheckpointComplete` fails, the task fails. The savepoint is still created, but the job ends in `FAILED` state: `StopWithSavepointTerminationHandlerImpl` calls `scheduler.handleGlobalFailure` when a task does not finish, and `StopWithSavepointStoppingException` is a `NonRecoverableError`. The stop-with-savepoint operation then fails visibly.
As a result, the two paths behave differently when creating the tag for the final savepoint fails:
| | Savepoint | Job | Tag | Stop-with-savepoint result |
| --- | --- | --- | --- | --- |
| Coordinator commit | Successful | `FINISHED` | Not created | Completes successfully |
| Operator commit | Successful | `FAILED` | Not created | Completes exceptionally |
This difference may be confusing for users.
I am working on a complete solution. The PoC validates the basic path, but completing and verifying consistent behavior across failover scenarios will take additional time. It may require synchronizing the task and coordinator so the final tag work is drained before termination. If the change grows beyond the scope of this PR, I may address it in a separate PR.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support savepoint auto-tag on coordinator commit path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #9309:
URL: https://github.com/apache/paimon/pull/9309
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support savepoint auto-tag on coordinator commit path [paimon]
Posted by "ifndef-SleePy (via GitHub)" <gi...@apache.org>.
ifndef-SleePy commented on code in PR #9309:
URL: https://github.com/apache/paimon/pull/9309#discussion_r3829904837
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Owns savepoint auto-tagging for {@link CommittingWriteOperatorCoordinator}, replicating the
+ * semantics of the classic {@link
+ * org.apache.paimon.flink.sink.AutoTagForSavepointCommitterOperator} for the coordinator-commit
+ * path. It keeps the set of savepoint checkpoint ids still awaiting a snapshot to tag; this set is
+ * deliberately not checkpointed but rebuilt from the savepoint ids replayed with each subtask's
+ * committables, so the coordinator's persisted state stays minimal.
+ */
+public class SavepointTagger {
+
+ private final SnapshotManager snapshotManager;
+ private final TagManager tagManager;
+ private final TagDeletion tagDeletion;
+ private final List<TagCallback> callbacks;
+ private final Duration tagTimeRetained;
+ // findSnapshotsForIdentifiers filters by commit user, so the tagger must be bound to the user
+ // the coordinator actually commits with (which the coordinator restores from its state).
+ private final String commitUser;
+ // Checkpoint ids of pending Flink savepoints awaiting a snapshot to tag.
+ private final NavigableSet<Long> pendingIdentifiers = new TreeSet<>();
+
+ public SavepointTagger(
+ SnapshotManager snapshotManager,
+ TagManager tagManager,
+ TagDeletion tagDeletion,
+ List<TagCallback> callbacks,
+ Duration tagTimeRetained,
+ String commitUser) {
+ this.snapshotManager = checkNotNull(snapshotManager);
+ this.tagManager = checkNotNull(tagManager);
+ this.tagDeletion = checkNotNull(tagDeletion);
+ this.callbacks = checkNotNull(callbacks);
+ this.tagTimeRetained = tagTimeRetained;
+ this.commitUser = checkNotNull(commitUser);
+ }
+
+ public void add(long savepointIdentifier) {
+ pendingIdentifiers.add(savepointIdentifier);
+ }
+
+ /**
+ * Tags every pending savepoint whose snapshot the commit up to {@code checkpointId} has
+ * materialized, then drops those pending intents.
+ */
+ public void tagUpTo(long checkpointId) {
+ NavigableSet<Long> headSet = pendingIdentifiers.headSet(checkpointId, true);
+ if (!headSet.isEmpty()) {
+ createTags(new ArrayList<>(headSet));
+ headSet.clear();
+ }
+ }
+
+ /** Drops an aborted savepoint's pending intent and removes any tag already created for it. */
+ public void dropAborted(long checkpointId) {
+ pendingIdentifiers.remove(checkpointId);
+ deleteTagIfExists(checkpointId);
+ }
+
+ private void createTags(Collection<Long> identifiers) {
+ List<Snapshot> snapshots =
+ snapshotManager.findSnapshotsForIdentifiers(
+ commitUser, new ArrayList<>(identifiers));
+ for (Snapshot snapshot : snapshots) {
+ String tagName = SavepointTagUtils.tagNameOf(snapshot.commitIdentifier());
+ // ignoreIfExists: a later checkpoint's completion may re-tag an already-tagged
+ // snapshot.
+ tagManager.createTag(snapshot, tagName, tagTimeRetained, callbacks, true);
+ }
+ }
+
+ private void deleteTagIfExists(long id) {
+ String tagName = SavepointTagUtils.tagNameOf(id);
+ if (tagManager.tagExists(tagName)) {
+ tagManager.deleteTag(tagName, tagDeletion, snapshotManager, callbacks);
Review Comment:
Nice catch. The operator-commit path has the same pre-existing behavior, so I fixed both paths. An aborted savepoint now deletes the tag only if its `commitUser` and `commitIdentifier` match the current commit user and the aborted checkpoint ID.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support savepoint auto-tag on coordinator commit path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9309:
URL: https://github.com/apache/paimon/pull/9309#discussion_r3817903306
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.sink.FlinkSinkBuilder;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
+import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.flink.util.AbstractTestBase;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end savepoint auto-tag tests for unaware-bucket append tables, parameterized over the two
+ * commit paths (coordinator-commit and the classic global committer). Both must create the same
+ * {@code savepoint-<checkpointId>} tag for a triggered savepoint.
+ *
+ * <p>Uses a source that emits continuously so the async savepoint deterministically lands on a
+ * data-carrying checkpoint; the empty-savepoint boundary (a separate, shared limitation) is
+ * intentionally avoided here.
+ */
+public class AppendTableSavepointTagITCase extends AbstractTestBase {
+
+ // The savepoint tag only materializes once a checkpoint *after* the savepoint completes and
+ // cumulatively commits the savepoint's snapshot (same catch-up as the classic path). Give the
+ // poll generous headroom so a transient checkpoint stall under load cannot trip the assertion.
+ private static final long WAIT_TIMEOUT_MILLIS = 120_000L;
+
+ @ParameterizedTest(name = "coordinatorCommit = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testSavepointCreatesTag(boolean coordinatorCommit) throws Exception {
+ String tableName = coordinatorCommit ? "T_COORD" : "T_CLASSIC";
+ FileStoreTable table = createTable(tableName, coordinatorCommit);
+
+ JobClient client = runSink(table);
+ try {
+ // Wait until a data-carrying snapshot exists so the async savepoint that follows
+ // deterministically lands on a checkpoint that carries data.
+ waitUntilSnapshotWithData(table);
+
+ client.triggerSavepoint(
+ getTempDirPath("savepoint_" + tableName), SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS);
+
+ // Poll until exactly one savepoint-prefixed tag appears, then assert it is consistent
+ // with the snapshot it points at.
+ Map<Snapshot, List<String>> savepointTags = waitUntilSavepointTagCreated(table);
+ assertThat(savepointTags).hasSize(1);
+ Map.Entry<Snapshot, List<String>> snapshotWithTags =
+ savepointTags.entrySet().iterator().next();
+ Snapshot tagged = snapshotWithTags.getKey();
+ assertThat(snapshotWithTags.getValue())
+ .containsExactly(SavepointTagUtils.tagNameOf(tagged.commitIdentifier()));
+ assertThat(table.snapshotManager().snapshotExists(tagged.id())).isTrue();
+ } finally {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * A sync savepoint (stop-with-savepoint) receives its own {@code notifyCheckpointComplete},
+ * unlike an async savepoint, so the tag is created for the savepoint's own snapshot rather than
+ * caught up by a later checkpoint. Both commit paths must still produce the same tag.
+ *
+ * <p>Disabled for now: on the coordinator-commit path this is racy. The coordinator creates the
+ * tag asynchronously on its single-thread commit executor (notifyCheckpointComplete ->
+ * tagUpTo), but stop-with-savepoint terminates the job right after the savepoint, and the
+ * coordinator's {@code close()} calls {@code commitExecutor.shutdownNow()}, which can drop the
+ * not-yet-run tag task so the tag is silently lost. The classic operator path is unaffected
+ * because it tags synchronously. Re-enable once the coordinator drains pending commit/tag work
+ * on end-of-input shutdown (the follow-up PR that adds proper end-input handling to the
+ * coordinator).
+ */
+ // TODO: enable once the coordinator supports end-input handling (drains pending tag work).
+ @Disabled(
Review Comment:
[P1] Please do not enable this configuration while stop-with-savepoint can lose its tag. `notifyCheckpointComplete` only queues commit/tag work on the coordinator executor, but job termination immediately calls `close()` and `shutdownNow()`, so a successful stop-with-savepoint can finish before `tagUpTo` runs. This test is disabled because it reproduces that exact race. Previously coordinator commit + auto-tag was rejected up front; after this PR users can select it and silently get a savepoint without the required Paimon tag. Please drain/await the queued commit/tag work during termination (or keep the precondition) and enable this test before advertising support.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Owns savepoint auto-tagging for {@link CommittingWriteOperatorCoordinator}, replicating the
+ * semantics of the classic {@link
+ * org.apache.paimon.flink.sink.AutoTagForSavepointCommitterOperator} for the coordinator-commit
+ * path. It keeps the set of savepoint checkpoint ids still awaiting a snapshot to tag; this set is
+ * deliberately not checkpointed but rebuilt from the savepoint ids replayed with each subtask's
+ * committables, so the coordinator's persisted state stays minimal.
+ */
+public class SavepointTagger {
+
+ private final SnapshotManager snapshotManager;
+ private final TagManager tagManager;
+ private final TagDeletion tagDeletion;
+ private final List<TagCallback> callbacks;
+ private final Duration tagTimeRetained;
+ // findSnapshotsForIdentifiers filters by commit user, so the tagger must be bound to the user
+ // the coordinator actually commits with (which the coordinator restores from its state).
+ private final String commitUser;
+ // Checkpoint ids of pending Flink savepoints awaiting a snapshot to tag.
+ private final NavigableSet<Long> pendingIdentifiers = new TreeSet<>();
+
+ public SavepointTagger(
+ SnapshotManager snapshotManager,
+ TagManager tagManager,
+ TagDeletion tagDeletion,
+ List<TagCallback> callbacks,
+ Duration tagTimeRetained,
+ String commitUser) {
+ this.snapshotManager = checkNotNull(snapshotManager);
+ this.tagManager = checkNotNull(tagManager);
+ this.tagDeletion = checkNotNull(tagDeletion);
+ this.callbacks = checkNotNull(callbacks);
+ this.tagTimeRetained = tagTimeRetained;
+ this.commitUser = checkNotNull(commitUser);
+ }
+
+ public void add(long savepointIdentifier) {
+ pendingIdentifiers.add(savepointIdentifier);
+ }
+
+ /**
+ * Tags every pending savepoint whose snapshot the commit up to {@code checkpointId} has
+ * materialized, then drops those pending intents.
+ */
+ public void tagUpTo(long checkpointId) {
+ NavigableSet<Long> headSet = pendingIdentifiers.headSet(checkpointId, true);
+ if (!headSet.isEmpty()) {
+ createTags(new ArrayList<>(headSet));
+ headSet.clear();
+ }
+ }
+
+ /** Drops an aborted savepoint's pending intent and removes any tag already created for it. */
+ public void dropAborted(long checkpointId) {
+ pendingIdentifiers.remove(checkpointId);
+ deleteTagIfExists(checkpointId);
+ }
+
+ private void createTags(Collection<Long> identifiers) {
+ List<Snapshot> snapshots =
+ snapshotManager.findSnapshotsForIdentifiers(
+ commitUser, new ArrayList<>(identifiers));
+ for (Snapshot snapshot : snapshots) {
+ String tagName = SavepointTagUtils.tagNameOf(snapshot.commitIdentifier());
+ // ignoreIfExists: a later checkpoint's completion may re-tag an already-tagged
+ // snapshot.
+ tagManager.createTag(snapshot, tagName, tagTimeRetained, callbacks, true);
+ }
+ }
+
+ private void deleteTagIfExists(long id) {
+ String tagName = SavepointTagUtils.tagNameOf(id);
+ if (tagManager.tagExists(tagName)) {
+ tagManager.deleteTag(tagName, tagDeletion, snapshotManager, callbacks);
Review Comment:
[P2] Only delete a tag when this coordinator observed/created the matching savepoint intent. `notifyCheckpointAborted` is invoked for normal checkpoints too, but `dropAborted` ignores whether the id was pending and unconditionally deletes the global `savepoint-<checkpointId>` name. Checkpoint ids restart in a fresh job, so aborting normal checkpoint N can delete a valid `savepoint-N` retained from an earlier job (and tag deletion may release its protected files). Please track ownership/created ids, or validate the tag snapshot/commit user before deleting it; add a test where a same-named pre-existing tag exists without a pending intent.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org