This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d0af6e841f [flink] Support savepoint auto-tag on coordinator commit
path (#9309)
d0af6e841f is described below
commit d0af6e841fd7168eca09b01b62572471e057be07
Author: Biao Liu <[email protected]>
AuthorDate: Sun Aug 23 23:28:55 2026 +0800
[flink] Support savepoint auto-tag on coordinator commit path (#9309)
---
.../sink/AutoTagForSavepointCommitterOperator.java | 14 +-
...dinatorCommittingRowDataStoreWriteOperator.java | 68 +-
.../org/apache/paimon/flink/sink/FlinkSink.java | 8 -
.../paimon/flink/sink/RowAppendTableSink.java | 62 +-
.../paimon/flink/sink/SavepointTagUtils.java | 67 ++
.../sink/coordinator/CheckpointCommittables.java | 32 +-
.../CheckpointCommittablesSerializer.java | 13 +-
.../CommittingWriteOperatorCoordinator.java | 68 +-
.../flink/sink/coordinator/SavepointTagger.java | 113 ++++
.../AppendTableSavepointTagFailoverITCase.java | 698 +++++++++++++++++++++
.../flink/AppendTableSavepointTagITCase.java | 248 ++++++++
.../AutoTagForSavepointCommitterOperatorTest.java | 41 +-
.../paimon/flink/sink/CommitterOperatorTest.java | 2 +-
...peratorTestBase.java => CommitterTestBase.java} | 6 +-
...torCommittingRowDataStoreWriteOperatorTest.java | 134 +++-
.../apache/paimon/flink/sink/FlinkSinkTest.java | 11 +-
.../paimon/flink/sink/SavepointTagUtilsTest.java | 132 ++++
.../CheckpointCommittablesSerializerTest.java | 37 +-
.../CommittingWriteOperatorCoordinatorTest.java | 173 ++++-
.../sink/coordinator/SavepointTaggerTest.java | 243 +++++++
20 files changed, 2104 insertions(+), 66 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
index 66d9781207..48f8d65e2e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java
@@ -54,7 +54,6 @@ import java.util.TreeSet;
*/
public class AutoTagForSavepointCommitterOperator<CommitT, GlobalCommitT>
implements OneInputStreamOperator<CommitT, CommitT>, BoundedOneInput {
- public static final String SAVEPOINT_TAG_PREFIX = "savepoint-";
private static final long serialVersionUID = 1L;
@@ -152,10 +151,13 @@ public class
AutoTagForSavepointCommitterOperator<CommitT, GlobalCommitT>
public void notifyCheckpointAborted(long checkpointId) throws Exception {
commitOperator.notifyCheckpointAborted(checkpointId);
identifiersForTags.remove(checkpointId);
- String tagName = SAVEPOINT_TAG_PREFIX + checkpointId;
- if (tagManager.tagExists(tagName)) {
- tagManager.deleteTag(tagName, tagDeletion, snapshotManager,
callbacks);
- }
+ SavepointTagUtils.deleteTagIfMatches(
+ tagManager,
+ commitOperator.getCommitUser(),
+ checkpointId,
+ tagDeletion,
+ snapshotManager,
+ callbacks);
}
private void createTagForIdentifiers(List<Long> identifiers) {
@@ -163,7 +165,7 @@ public class AutoTagForSavepointCommitterOperator<CommitT,
GlobalCommitT>
snapshotManager.findSnapshotsForIdentifiers(
commitOperator.getCommitUser(), identifiers);
for (Snapshot snapshot : snapshotForTags) {
- String tagName = SAVEPOINT_TAG_PREFIX +
snapshot.commitIdentifier();
+ String tagName =
SavepointTagUtils.tagNameOf(snapshot.commitIdentifier());
// shouldn't throw exception when tag exists
tagManager.createTag(snapshot, tagName, tagTimeRetained,
callbacks, true);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java
index 3bd617b5eb..7fcb87d15b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java
@@ -33,9 +33,11 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.StateInitializationContext;
-import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -71,6 +73,9 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
private final OperatorEventGateway operatorEventGateway;
+ /** Whether savepoint auto-tagging is enabled; when off the writer never
flags a tag intent. */
+ private final boolean autoTagForSavepoint;
+
/** Persisted buffer of pending checkpoints not yet acknowledged by the
coordinator. */
private transient ListState<CheckpointCommittables>
pendingCommittableState;
@@ -97,9 +102,11 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
FileStoreTable table,
StoreSinkWrite.Provider storeSinkWriteProvider,
String initialCommitUser,
- OperatorEventGateway operatorEventGateway) {
+ OperatorEventGateway operatorEventGateway,
+ boolean autoTagForSavepoint) {
super(parameters, table, storeSinkWriteProvider, initialCommitUser);
this.operatorEventGateway =
Preconditions.checkNotNull(operatorEventGateway);
+ this.autoTagForSavepoint = autoTagForSavepoint;
}
@Override
@@ -152,10 +159,26 @@ public class
CoordinatorCommittingRowDataStoreWriteOperator
}
@Override
- public void snapshotState(StateSnapshotContext context) throws Exception {
- super.snapshotState(context);
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation)
+ throws Exception {
+ // Ordering within a checkpoint: emitCommittables already ran (in
prepareSnapshotPreBarrier,
+ // before the barrier) and buffered this checkpoint's committables
into pendingCommittables.
+ if (autoTagForSavepoint &&
checkpointOptions.getCheckpointType().isSavepoint()) {
+ pendingCommittables.computeIfPresent(
+ checkpointId,
+ (id, checkpointCommittables) ->
+
checkpointCommittables.withShouldCreateSavepointTag(true));
+ }
+ // Report here, not in emitCommittables, so the savepoint-tag intent
is known before
+ // sending.
+ reportToCoordinator(checkpointId);
pendingCommittableState.clear();
pendingCommittableState.addAll(new
ArrayList<>(pendingCommittables.values()));
+ return super.snapshotState(checkpointId, timestamp, checkpointOptions,
storageLocation);
}
@Override
@@ -166,15 +189,27 @@ public class
CoordinatorCommittingRowDataStoreWriteOperator
pendingCommittables.headMap(checkpointId, true).clear();
}
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ super.notifyCheckpointAborted(checkpointId);
+ if (!autoTagForSavepoint) {
+ return;
+ }
+ // Drop only the savepoint-tag intent on the aborted committables
(keep the data). A later
+ // checkpoint must not persist a stale intent, or the coordinator
would recreate a tag for a
+ // gone savepoint. Mirrors the operator path pruning the aborted id.
+ pendingCommittables.computeIfPresent(
+ checkpointId,
+ (id, checkpointCommittables) ->
+
checkpointCommittables.withShouldCreateSavepointTag(false));
+ }
+
@Override
protected void emitCommittables(boolean waitCompaction, long checkpointId)
throws IOException {
List<Committable> committables = prepareCommit(waitCompaction,
checkpointId);
CheckpointCommittables entry =
new CheckpointCommittables(
checkpointId, committables, currentWatermark,
currentIdle);
- // Emit an event per (subtask, checkpoint) regardless of whether
committables is empty.
- operatorEventGateway.sendEventToCoordinator(
- CommittableEvent.create(checkpointId, entry, eventSerializer));
// Always buffer the per-checkpoint entry so an empty barrier — even
one that has not seen
// a real watermark yet — survives restore. The coordinator relies on
every subtask
// having an entry for the checkpoint being aligned so its watermark
min stays sound.
@@ -185,6 +220,25 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
committables.forEach(committable -> output.collect(new
StreamRecord<>(committable)));
}
+ @Override
+ public void endInput() throws Exception {
+ super.endInput();
+ // endInput emits the Long.MAX_VALUE committables but is not followed
by a snapshotState,
+ // so report them here just to keep the existing behavior.
+ // TODO: revisit how end-of-input committables should be handled.
+ reportToCoordinator(Long.MAX_VALUE);
+ }
+
+ /**
+ * Sends the buffered committables for {@code checkpointId} to the
coordinator, one per
+ * checkpoint.
+ */
+ private void reportToCoordinator(long checkpointId) throws IOException {
+ operatorEventGateway.sendEventToCoordinator(
+ CommittableEvent.create(
+ checkpointId, pendingCommittables.get(checkpointId),
eventSerializer));
+ }
+
@Override
public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 5d00584606..1445061bfc 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -410,14 +410,6 @@ public abstract class FlinkSink<T> implements Serializable
{
+ PRECOMMIT_COMPACT.key()
+ " = false.");
- // The OperatorCoordinator cannot tell a savepoint from a normal
checkpoint.
- // TODO support savepoint auto-tag.
- checkArgument(
- !options.get(SINK_AUTO_TAG_FOR_SAVEPOINT),
- "Could not enable coordinator commit because "
- + SINK_AUTO_TAG_FOR_SAVEPOINT.key()
- + " is enabled, which is not supported yet.");
-
// TODO concurrent checkpoints are not supported yet.
checkArgument(
checkpointConfig.getMaxConcurrentCheckpoints() == 1,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
index 443c9b6769..2e6b54fef1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java
@@ -21,9 +21,15 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
import
org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator;
+import org.apache.paimon.flink.sink.coordinator.SavepointTagger;
import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.SerializableSupplier;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
@@ -33,6 +39,8 @@ import
org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import java.time.Duration;
+import java.util.List;
import java.util.Map;
/** An {@link AppendTableSink} which handles {@link InternalRow}. */
@@ -56,7 +64,9 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
// checkpointing on by default for the JM-side committer;
bounded sources will
// be handled by end-input support in a follow-up PR
true,
- createCommitterFactory());
+ createCommitterFactory(),
+ new Options(table.options())
+
.get(FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT));
}
return createNoStateRowWriteOperatorFactory(table, writeProvider,
commitUser);
}
@@ -82,9 +92,15 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
StoreSinkWrite.Provider writeProvider,
String commitUser,
boolean streamingCheckpointEnabled,
- Committer.Factory<Committable, ManifestCommittable>
committerFactory) {
+ Committer.Factory<Committable, ManifestCommittable>
committerFactory,
+ boolean autoTagForSavepoint) {
return new CoordinatorCommittingFactory(
- table, writeProvider, commitUser, streamingCheckpointEnabled,
committerFactory);
+ table,
+ writeProvider,
+ commitUser,
+ streamingCheckpointEnabled,
+ committerFactory,
+ autoTagForSavepoint);
}
private static class CoordinatorCommittingFactory extends
RowDataStoreWriteOperator.Factory
@@ -94,23 +110,52 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
private final boolean streamingCheckpointEnabled;
private final Committer.Factory<Committable, ManifestCommittable>
committerFactory;
+ private final boolean autoTagForSavepoint;
CoordinatorCommittingFactory(
FileStoreTable table,
StoreSinkWrite.Provider storeSinkWriteProvider,
String initialCommitUser,
boolean streamingCheckpointEnabled,
- Committer.Factory<Committable, ManifestCommittable>
committerFactory) {
+ Committer.Factory<Committable, ManifestCommittable>
committerFactory,
+ boolean autoTagForSavepoint) {
super(table, storeSinkWriteProvider, initialCommitUser);
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.committerFactory = committerFactory;
+ this.autoTagForSavepoint = autoTagForSavepoint;
}
@Override
public OperatorCoordinator.Provider getCoordinatorProvider(
String operatorName, OperatorID operatorID) {
return new CommittingWriteOperatorCoordinator.Provider(
- operatorID, committerFactory, streamingCheckpointEnabled,
initialCommitUser);
+ operatorID,
+ committerFactory,
+ streamingCheckpointEnabled,
+ initialCommitUser,
+ autoTagForSavepoint ? createSavepointTaggerFactory(table)
: null);
+ }
+
+ /**
+ * Builds the savepoint auto-tag factory. The factory captures only
serializable suppliers
+ * and binds the commit user late, on the JM, once the coordinator has
restored it.
+ */
+ private static SavepointTagger.Factory
createSavepointTaggerFactory(FileStoreTable table) {
+ SerializableSupplier<SnapshotManager> snapshotManagerFactory =
table::snapshotManager;
+ SerializableSupplier<TagManager> tagManagerFactory =
table::tagManager;
+ SerializableSupplier<TagDeletion> tagDeletionFactory =
+ () -> table.store().newTagDeletion();
+ SerializableSupplier<List<TagCallback>> callbacksSupplier =
+ () -> table.store().createTagCallbacks(table);
+ Duration tagTimeRetained =
table.coreOptions().tagDefaultTimeRetained();
+ return commitUser ->
+ new SavepointTagger(
+ snapshotManagerFactory.get(),
+ tagManagerFactory.get(),
+ tagDeletionFactory.get(),
+ callbacksSupplier.get(),
+ tagTimeRetained,
+ commitUser);
}
@Override
@@ -122,7 +167,12 @@ public class RowAppendTableSink extends
AppendTableSink<InternalRow> {
parameters.getOperatorEventDispatcher().getOperatorEventGateway(operatorId);
return (T)
new CoordinatorCommittingRowDataStoreWriteOperator(
- parameters, table, storeSinkWriteProvider,
initialCommitUser, gateway);
+ parameters,
+ table,
+ storeSinkWriteProvider,
+ initialCommitUser,
+ gateway,
+ autoTagForSavepoint);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
new file mode 100644
index 0000000000..b530cd864c
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SavepointTagUtils.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.tag.Tag;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.util.List;
+
+/** Helpers for savepoint tags — the tags Paimon auto-creates to mark Flink
savepoints. */
+public class SavepointTagUtils {
+
+ /** Prefix shared by every savepoint auto-tag; use {@link
#tagNameOf(long)} for a full name. */
+ public static final String PREFIX = "savepoint-";
+
+ /** Name of the auto-tag for a savepoint committed under {@code
commitIdentifier}. */
+ public static String tagNameOf(long commitIdentifier) {
+ return PREFIX + commitIdentifier;
+ }
+
+ /**
+ * Whether the tag belongs to the savepoint committed by the given user
and identifier.
+ *
+ * <p>Checkpoint identifiers restart with each job, so a matching tag name
alone does not prove
+ * ownership.
+ */
+ public static boolean isSavepointTagFor(Tag tag, String commitUser, long
commitIdentifier) {
+ return commitUser.equals(tag.commitUser()) && tag.commitIdentifier()
== commitIdentifier;
+ }
+
+ /** Deletes the savepoint tag if it belongs to the given commit user and
identifier. */
+ public static void deleteTagIfMatches(
+ TagManager tagManager,
+ String commitUser,
+ long commitIdentifier,
+ TagDeletion tagDeletion,
+ SnapshotManager snapshotManager,
+ List<TagCallback> callbacks) {
+ String tagName = tagNameOf(commitIdentifier);
+ tagManager
+ .get(tagName)
+ .filter(tag -> isSavepointTagFor(tag, commitUser,
commitIdentifier))
+ .ifPresent(
+ ignored ->
+ tagManager.deleteTag(
+ tagName, tagDeletion, snapshotManager,
callbacks));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
index 3e08c1f77f..98cdd18291 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
@@ -35,19 +35,33 @@ public class CheckpointCommittables {
// Idle bit is frozen at barrier time together with watermark; mirrors
what Flink's
// StatusWatermarkValve would have observed on the writer's input at the
moment of the barrier.
private final boolean idle;
+ // Whether a savepoint tag should be created for the checkpoint that
produced these
+ // committables.
+ private final boolean shouldCreateSavepointTag;
public CheckpointCommittables(
- long checkpointId, List<Committable> committables, long watermark,
boolean idle) {
+ long checkpointId,
+ List<Committable> committables,
+ long watermark,
+ boolean idle,
+ boolean shouldCreateSavepointTag) {
this.checkpointId = checkpointId;
this.committables = committables;
this.watermark = watermark;
this.idle = idle;
+ this.shouldCreateSavepointTag = shouldCreateSavepointTag;
+ }
+
+ // Convenience for callers that are not savepoint-aware yet.
+ public CheckpointCommittables(
+ long checkpointId, List<Committable> committables, long watermark,
boolean idle) {
+ this(checkpointId, committables, watermark, idle, false);
}
// Convenience for callers that only need the pre-idle-aware shape (ACTIVE
writer).
public CheckpointCommittables(
long checkpointId, List<Committable> committables, long watermark)
{
- this(checkpointId, committables, watermark, false);
+ this(checkpointId, committables, watermark, false, false);
}
public long checkpointId() {
@@ -66,6 +80,16 @@ public class CheckpointCommittables {
return idle;
}
+ public boolean shouldCreateSavepointTag() {
+ return shouldCreateSavepointTag;
+ }
+
+ /** Returns a copy with the savepoint-tag intent set; all other fields
preserved. */
+ public CheckpointCommittables withShouldCreateSavepointTag(boolean
shouldCreateSavepointTag) {
+ return new CheckpointCommittables(
+ checkpointId, committables, watermark, idle,
shouldCreateSavepointTag);
+ }
+
public int size() {
return committables.size();
}
@@ -77,7 +101,7 @@ public class CheckpointCommittables {
@Override
public String toString() {
return String.format(
- "CheckpointCommittables{checkpointId=%d, watermark=%d,
idle=%s, committables=%s}",
- checkpointId, watermark, idle, committables);
+ "CheckpointCommittables{checkpointId=%d, watermark=%d,
idle=%s, shouldCreateSavepointTag=%s, committables=%s}",
+ checkpointId, watermark, idle, shouldCreateSavepointTag,
committables);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
index b557559996..5e51385c80 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
@@ -43,7 +43,8 @@ public class CheckpointCommittablesSerializer
public int getVersion() {
// v1: checkpointId + watermark + committables
// v2: v1 + idle bit (appended before the committable list to keep the
ordering explicit)
- return 2;
+ // v3: v2 + savepoint-tag bit (appended after idle, still before the
committable list)
+ return 3;
}
@Override
@@ -52,6 +53,7 @@ public class CheckpointCommittablesSerializer
out.writeLong(value.checkpointId());
out.writeLong(value.watermark());
out.writeBoolean(value.idle());
+ out.writeBoolean(value.shouldCreateSavepointTag());
// Nested serializer version comes before the list so the reader can
pick the right decoder
// before touching any list bytes — mirrors
ManifestCommittableSerializer's layout.
out.writeInt(committableSerializer.getVersion());
@@ -67,7 +69,7 @@ public class CheckpointCommittablesSerializer
@Override
public CheckpointCommittables deserialize(int version, byte[] serialized)
throws IOException {
- if (version != 1 && version != 2) {
+ if (version != 1 && version != 2 && version != 3) {
throw new IOException("Unknown version " + version);
}
DataInputDeserializer in = new DataInputDeserializer(serialized);
@@ -75,7 +77,9 @@ public class CheckpointCommittablesSerializer
long watermark = in.readLong();
// v1 payloads pre-date idle tracking; default to ACTIVE (idle=false),
which reproduces
// the pre-idle-aware behaviour: every subtask contributes to the min
unconditionally.
- boolean idle = version >= 2 && in.readBoolean();
+ boolean idle = version >= 2 ? in.readBoolean() : false;
+ // v1/v2 payloads pre-date the savepoint-tag bit; older payloads have
no tag.
+ boolean shouldCreateSavepointTag = version >= 3 ? in.readBoolean() :
false;
int committableVersion = in.readInt();
int count = in.readInt();
List<Committable> committables = new ArrayList<>(count);
@@ -85,6 +89,7 @@ public class CheckpointCommittablesSerializer
in.readFully(bytes);
committables.add(committableSerializer.deserialize(committableVersion, bytes));
}
- return new CheckpointCommittables(checkpointId, committables,
watermark, idle);
+ return new CheckpointCommittables(
+ checkpointId, committables, watermark, idle,
shouldCreateSavepointTag);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
index 3ef4b26d42..e792432a31 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
@@ -39,6 +39,8 @@ import org.apache.flink.util.function.ThrowingRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -77,6 +79,7 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
private final Committer.Factory<Committable, ManifestCommittable>
committerFactory;
private final boolean streamingCheckpointEnabled;
private final int parallelism;
+ @Nullable private final SavepointTagger.Factory savepointTaggerFactory;
private final WriterCommittables[] subtaskCommittables;
private final TypeSerializer<CheckpointCommittables>
committablesSerializer;
@@ -95,16 +98,20 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
private Committer<Committable, ManifestCommittable> committer;
private String commitUser;
private MemoryBackendStateStore stateStore;
+ // Built in initializeAfterRestore once commitUser is known; null when
auto-tag is disabled.
+ @Nullable private SavepointTagger savepointTagger;
public CommittingWriteOperatorCoordinator(
OperatorCoordinator.Context context,
Committer.Factory<Committable, ManifestCommittable>
committerFactory,
boolean streamingCheckpointEnabled,
- String initialCommitUser) {
+ String initialCommitUser,
+ @Nullable SavepointTagger.Factory savepointTaggerFactory) {
this.context = context;
this.committerFactory = committerFactory;
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.commitUser = initialCommitUser;
+ this.savepointTaggerFactory = savepointTaggerFactory;
this.parallelism = context.currentParallelism();
this.subtaskCommittables = new WriterCommittables[parallelism];
this.committablesSerializer =
@@ -134,11 +141,11 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
restoreState(restoredCheckpointId,
restoredCheckpointData);
// not needed after deserialization; release the
reference
restoredCheckpointData = null;
- initializeCommitter(true);
+ initializeAfterRestore(true);
// stay in RESTORING until writers re-emit
committables and align catches up
} else {
restoreState(OperatorCoordinator.NO_CHECKPOINT, null);
- initializeCommitter(false);
+ initializeAfterRestore(false);
transitionState(State.RUNNING);
}
},
@@ -233,11 +240,33 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
throw new RuntimeException(e);
}
});
+ // An async savepoint does not fire
notifyCheckpointComplete for its own id
+ // (FLIP-193), so its tag cannot be created when the
savepoint completes.
+ // Catch up on each checkpoint completion instead, tagging
every pending
+ // savepoint id up to checkpointId once the commit
materialized its snapshot.
+ if (savepointTagger != null) {
+ savepointTagger.tagUpTo(checkpointId);
+ }
},
"completing checkpoint %d",
checkpointId);
}
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) {
+ // Runs tag I/O on the commit executor, never the JM main thread. An
aborted savepoint may
+ // already have been tagged by a later checkpoint's completion
(cumulative commit), so drop
+ // the pending intent and remove any tag that was created.
+ runInEventLoop(
+ () -> {
+ if (savepointTagger != null) {
+ savepointTagger.dropAborted(checkpointId);
+ }
+ },
+ "aborting checkpoint %d",
+ checkpointId);
+ }
+
/**
* Called by the framework at most once, before {@link #start()}. May be
skipped entirely if the
* job has no checkpoint or savepoint to restore from; in that case the
coordinator goes
@@ -333,6 +362,16 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
}
private void updateSubtaskCommittables(int subtask, WriterCommittables
incoming) {
+ if (savepointTagger != null) {
+ // Collect savepoint intents as events arrive (steady state and
restore both funnel
+ // here), rebuilding the pending-tag set without checkpointing it.
+ for (CheckpointCommittables checkpointCommittables :
+ incoming.getCommittablesPerCheckpoint().values()) {
+ if (checkpointCommittables.shouldCreateSavepointTag()) {
+ savepointTagger.add(checkpointCommittables.checkpointId());
+ }
+ }
+ }
if (subtaskCommittables[subtask] != null) {
subtaskCommittables[subtask].mergeWith(incoming);
} else {
@@ -360,6 +399,11 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
checkpointId, subtaskCommittables,
watermarkPerCheckpoint, committer),
watermarkPerCheckpoint,
committables -> committer.filterAndCommit(committables, true,
true));
+ // Tag any restored savepoint(s) whose snapshot the re-commit
materialized, so a
+ // restore-from-savepoint still produces the savepoint tag.
+ if (savepointTagger != null) {
+ savepointTagger.tagUpTo(checkpointId);
+ }
}
@VisibleForTesting
@@ -471,7 +515,7 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
}
}
- private void initializeCommitter(boolean isRestored) {
+ private void initializeAfterRestore(boolean isRestored) {
// Coordinator runs at parallelism 1 (single instance per JobVertex),
matching
// CommitterOperator's contract; hardcode parallelism=1 /
subtaskIndex=0
Committer.Context committerContext =
@@ -484,6 +528,11 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
1,
0);
committer = committerFactory.create(committerContext);
+ // Bind the tagger to the (possibly restored) commit user, so
findSnapshotsForIdentifiers
+ // matches the snapshots this coordinator commits.
+ if (savepointTaggerFactory != null) {
+ savepointTagger = savepointTaggerFactory.create(commitUser);
+ }
}
private void transitionState(State targetState) {
@@ -600,22 +649,29 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
private final Committer.Factory<Committable, ManifestCommittable>
committerFactory;
private final boolean streamingCheckpointEnabled;
private final String initialCommitUser;
+ @Nullable private final SavepointTagger.Factory savepointTaggerFactory;
public Provider(
OperatorID operatorId,
Committer.Factory<Committable, ManifestCommittable>
committerFactory,
boolean streamingCheckpointEnabled,
- String initialCommitUser) {
+ String initialCommitUser,
+ @Nullable SavepointTagger.Factory savepointTaggerFactory) {
super(operatorId);
this.committerFactory = committerFactory;
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.initialCommitUser = initialCommitUser;
+ this.savepointTaggerFactory = savepointTaggerFactory;
}
@Override
public OperatorCoordinator getCoordinator(OperatorCoordinator.Context
context) {
return new CommittingWriteOperatorCoordinator(
- context, committerFactory, streamingCheckpointEnabled,
initialCommitUser);
+ context,
+ committerFactory,
+ streamingCheckpointEnabled,
+ initialCommitUser,
+ savepointTaggerFactory);
}
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
new file mode 100644
index 0000000000..e762f55e01
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java
@@ -0,0 +1,113 @@
+/*
+ * 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);
+ SavepointTagUtils.deleteTagIfMatches(
+ tagManager, commitUser, checkpointId, tagDeletion,
snapshotManager, callbacks);
+ }
+
+ 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);
+ }
+ }
+
+ /** Builds a {@link SavepointTagger} bound to the coordinator's restored
commit user. */
+ public interface Factory extends Serializable {
+ SavepointTagger create(String commitUser);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java
new file mode 100644
index 0000000000..49198b4f51
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagFailoverITCase.java
@@ -0,0 +1,698 @@
+/*
+ * 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.manifest.ManifestCommittable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitCallback;
+
+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.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
+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.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+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.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.util.ExceptionUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Failover and restore behavior of savepoint auto-tag for unaware-bucket
append tables, asserting
+ * that the coordinator-commit path stays at parity with the classic
operator-commit path.
+ */
+public class AppendTableSavepointTagFailoverITCase extends AbstractTestBase {
+
+ private static final long WAIT_TIMEOUT_MILLIS = 120_000L;
+
+ @BeforeEach
+ public void resetInjectors() {
+ // These injectors coordinate through static fields shared across
every test in this JVM;
+ // reset them up front so no test inherits another's leftover
arming/one-shot state.
+ FailOnSavepointOperator.reset();
+ BlockCheckpointAfterSavepointOperator.disarmBlocking();
+ FailOnFirstPostRecoveryCommitCallback.reset();
+ }
+
+ /** The tag already exists on restore; recover() must re-tag idempotently.
*/
+ @ParameterizedTest(name = "coordinatorCommit = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testSavepointTagIdempotentOnRestore(boolean coordinatorCommit)
throws Exception {
+ String tableName = coordinatorCommit ? "T_COORD_IDEMPOTENT" :
"T_CLASSIC_IDEMPOTENT";
+ FileStoreTable table = createTable(tableName, coordinatorCommit);
+
+ // Phase 1: run the job, take a savepoint, and let its tag
materialize. Keep the tag in
+ // place so phase 2 restores into a state where the tag already exists.
+ String savepointPath;
+ long taggedIdentifier;
+ JobClient firstClient = runSink(table, null);
+ try {
+ waitUntilSnapshotWithData(table);
+ savepointPath =
+ firstClient
+ .triggerSavepoint(
+ getTempDirPath("savepoint_" + tableName),
+ SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS);
+ Map<Snapshot, List<String>> tags =
waitUntilSavepointTagCreated(table);
+ assertThat(tags).hasSize(1);
+ taggedIdentifier =
tags.keySet().iterator().next().commitIdentifier();
+ } finally {
+ firstClient.cancel().get(30, TimeUnit.SECONDS);
+ }
+
+ // Phase 2: restore from the savepoint while the tag is already
present. recover() re-runs
+ // the tag creation; SavepointTagger.createTag with ignoreIfExists
must make it a no-op, so
+ // the tag stays unique and unchanged.
+ JobClient secondClient = runSink(table, savepointPath);
+ try {
+ // A snapshot committed after restore proves recover() has already
run (its re-tag
+ // happens before the resumed job commits again), so no fixed
sleep is needed.
+ waitUntilRecoveredAndCommitting(table);
+ Map<Snapshot, List<String>> tags = savepointTags(table);
+ assertThat(tags).hasSize(1);
+ Map.Entry<Snapshot, List<String>> snapshotWithTags =
tags.entrySet().iterator().next();
+ assertThat(snapshotWithTags.getValue())
+ .containsExactly(
+ SavepointTagUtils.tagNameOf(
+
snapshotWithTags.getKey().commitIdentifier()));
+
assertThat(snapshotWithTags.getKey().commitIdentifier()).isEqualTo(taggedIdentifier);
+ } finally {
+ secondClient.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * The tag was never created (no checkpoint completed after the
savepoint); restore must create
+ * it.
+ */
+ @ParameterizedTest(name = "coordinatorCommit = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testSavepointTagRecreatedOnRestore(boolean coordinatorCommit)
throws Exception {
+ String tableName = coordinatorCommit ? "T_COORD_RECREATE" :
"T_CLASSIC_RECREATE";
+ FileStoreTable table = createTable(tableName, coordinatorCommit);
+
+ // Phase 1: take a savepoint, and block every normal checkpoint that
would follow it so none
+ // can materialize the tag before we cancel. The savepoint's snapshot
is captured but its
+ // tag
+ // is never created yet.
+ BlockCheckpointAfterSavepointOperator.armBlocking();
+ String savepointPath;
+ JobClient firstClient = runSink(table, null, true);
+ try {
+ waitUntilSnapshotWithData(table);
+ savepointPath =
+ firstClient
+ .triggerSavepoint(
+ getTempDirPath("savepoint_" + tableName),
+ SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS);
+ } finally {
+ firstClient.cancel().get(30, TimeUnit.SECONDS);
+ }
+ assertThat(savepointTags(table)).isEmpty();
+
+ // Phase 2: restore from the savepoint with blocking disarmed (same
topology so the writer
+ // state maps back). The writer replays the pending savepoint bit; the
commit that follows
+ // materializes the savepoint's snapshot and its tag is created for
the first time here (via
+ // recover() on coordinator, or committer ListState on the classic
path).
+ BlockCheckpointAfterSavepointOperator.disarmBlocking();
+ JobClient secondClient = runSink(table, savepointPath, true);
+ try {
+ Map<Snapshot, List<String>> tags =
waitUntilSavepointTagCreated(table);
+ assertThat(tags).hasSize(1);
+ Map.Entry<Snapshot, List<String>> snapshotWithTags =
tags.entrySet().iterator().next();
+ assertThat(snapshotWithTags.getValue())
+ .containsExactly(
+ SavepointTagUtils.tagNameOf(
+
snapshotWithTags.getKey().commitIdentifier()));
+ } finally {
+ secondClient.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * Region failover: a single writer subtask throws while a savepoint is in
flight, so only that
+ * region restarts and the coordinator keeps running. The interrupted
savepoint never commits
+ * its snapshot, so it produces no tag; a savepoint taken after recovery
must be tagged. Region
+ * failover is a coordinator-commit-only concern (the classic committer
shares a failover region
+ * with the writers), so this is not parameterized.
+ */
+ @Test
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testRegionFailoverPreservesSavepointTag() throws Exception {
+ String tableName = "T_COORD_REGION_FAILOVER";
+ FileStoreTable table = createTable(tableName, true);
+
+ Configuration conf = new Configuration();
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
Integer.MAX_VALUE);
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
Duration.ofSeconds(1));
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.setParallelism(2);
+ env.enableCheckpointing(200);
+ DataStreamSource<RowData> source =
+ env.fromSource(
+ new ContinuousSource(),
WatermarkStrategy.noWatermarks(), "region-source");
+ // forward + same parallelism so source, fail-injector, and writer are
one failover region
+ // per subtask; a single subtask's failure restarts only its region,
not the whole job.
+ DataStream<RowData> injected =
+ source.forward()
+ .transform(
+ "fail-on-savepoint",
+ InternalTypeInfo.of(RowType.of(new IntType(),
new VarCharType())),
+ new FailOnSavepointOperator(1))
+ .setParallelism(2);
+ new FlinkSinkBuilder(table).forRowData(injected).build();
+ JobClient client = env.executeAsync("region-failover-savepoint-tag");
+ try {
+ waitUntilSnapshotWithData(table);
+
+ // Savepoint 1: subtask 1 throws while snapshotting it, which
aborts the in-flight
+ // savepoint and triggers a region failover. The trigger future
must fail, and the
+ // failure must be the savepoint being interrupted by that
failover — the checkpoint
+ // coordinator suspends — not some unrelated error. The injected
exception is not on
+ // this
+ // chain: it fails the task (driving the failover) rather than
propagating to the
+ // trigger
+ // future, which only sees the coordinator suspending, so
CheckpointException is the
+ // tightest type we can assert here.
+ assertThatThrownBy(
+ () ->
+ client.triggerSavepoint(
+
getTempDirPath("savepoint1_" + tableName),
+
SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS))
+ .satisfies(
+ e ->
+ assertThat(
+
ExceptionUtils.findThrowable(
+ e,
CheckpointException.class))
+ .isPresent());
+ assertThat(savepointTags(table)).isEmpty();
+
+ // Wait until the job has recovered and resumed committing after
the region failover.
+ waitUntilRecoveredAndCommitting(table);
+
+ // Savepoint 2: after recovery it must be tagged correctly,
proving the region failover
+ // did not break the coordinator's auto-tag state.
+ client.triggerSavepoint(
+ getTempDirPath("savepoint2_" + tableName),
SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS);
+ Map<Snapshot, List<String>> tags =
waitUntilSavepointTagCreated(table);
+ assertThat(tags).hasSize(1);
+ Map.Entry<Snapshot, List<String>> snapshotWithTags =
tags.entrySet().iterator().next();
+ assertThat(snapshotWithTags.getValue())
+ .containsExactly(
+ SavepointTagUtils.tagNameOf(
+
snapshotWithTags.getKey().commitIdentifier()));
+
assertThat(table.snapshotManager().snapshotExists(snapshotWithTags.getKey().id()))
+ .isTrue();
+ } finally {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ /**
+ * An async savepoint is aborted (one writer failed while others
succeeded), then a global
+ * failover restores from the first normal checkpoint after the abort. The
surviving writer
+ * still carries the aborted savepoint's bit in that checkpoint's state,
so the coordinator must
+ * NOT recreate the savepoint's tag on restore. This is the
coordinator-vs-operator consistency
+ * gap: the operator path prunes the aborted id from its checkpointed set,
while the coordinator
+ * rebuilds its pending-tag set from the writer-replayed bits.
+ *
+ * <p>Deterministic reproduction of "restore exactly from the one
checkpoint that carries the
+ * stale bit":
+ *
+ * <ol>
+ * <li>Parallelism 2. A savepoint S is taken; subtask 1 throws in its
snapshot, aborting S and
+ * forcing a region failover while the coordinator (and subtask 0)
keep running. Subtask 0
+ * keeps S's savepoint bit in its in-memory pending buffer (the
writer has no
+ * notifyCheckpointAborted hook to clear it).
+ * <li>The job resumes and the first normal checkpoint C after recovery
persists subtask 0's
+ * buffer — including S's stale bit — into C's operator state.
+ * <li>A commit callback throws exactly once, on the commit of that
first post-recovery
+ * checkpoint, to force a GLOBAL failover whose latest completed
checkpoint is C.
+ * <li>Global restore replays C's committables (with S's stale bit) to a
fresh coordinator,
+ * which re-collects S and, after recover() commits S's snapshot,
would tag it. The tag is
+ * an orphan: S was aborted and its Flink savepoint no longer exists.
+ * </ol>
+ *
+ * <p>The assertion is that no savepoint tag exists after the dust
settles. This is
+ * coordinator-commit-only (region failover requires the coordinator
path), so it is not
+ * parameterized.
+ */
+ @Test
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testAbortedSavepointNotRetaggedAfterGlobalFailover() throws
Exception {
+ String tableName = "T_COORD_ABORT_RETAG";
+ FileStoreTable table =
+ createTable(tableName, true,
FailOnFirstPostRecoveryCommitCallback.class.getName());
+
+ Configuration conf = new Configuration();
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
Integer.MAX_VALUE);
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
Duration.ofSeconds(1));
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.setParallelism(2);
+ env.enableCheckpointing(200);
+ DataStreamSource<RowData> source =
+ env.fromSource(
+ new ContinuousSource(),
WatermarkStrategy.noWatermarks(), "retag-source");
+ // forward + same parallelism so each source/fail-injector/writer
triple is its own failover
+ // region; a single subtask's failure restarts only its region,
keeping the coordinator up.
+ DataStream<RowData> injected =
+ source.forward()
+ .transform(
+ "fail-on-savepoint",
+ InternalTypeInfo.of(RowType.of(new IntType(),
new VarCharType())),
+ new FailOnSavepointOperator(1))
+ .setParallelism(2);
+ new FlinkSinkBuilder(table).forRowData(injected).build();
+ JobClient client = env.executeAsync("aborted-savepoint-retag");
+ try {
+ waitUntilSnapshotWithData(table);
+
+ // Savepoint S: subtask 1 throws while snapshotting it, which
aborts S and triggers a
+ // region failover; subtask 0 survives with S's savepoint bit
still buffered. The
+ // trigger
+ // future must fail with the savepoint being interrupted by that
failover (the
+ // checkpoint
+ // coordinator suspends), not some unrelated error. The injected
exception is not on
+ // this
+ // chain: it fails the task rather than propagating to the trigger
future, so
+ // CheckpointException is the tightest type we can assert here.
+ assertThatThrownBy(
+ () ->
+ client.triggerSavepoint(
+
getTempDirPath("savepoint_" + tableName),
+
SavepointFormatType.DEFAULT)
+ .get(60, TimeUnit.SECONDS))
+ .satisfies(
+ e ->
+ assertThat(
+
ExceptionUtils.findThrowable(
+ e,
CheckpointException.class))
+ .isPresent());
+ assertThat(savepointTags(table)).isEmpty();
+
+ // The commit callback throws once on the first post-recovery
commit, forcing a global
+ // failover whose latest completed checkpoint C carries subtask
0's stale savepoint bit.
+ waitUntilCallbackFired();
+ // After the global failover the coordinator restores from C,
replays the stale bit, and
+ // (before the fix) would recreate the orphan tag during
recover(). Wait for two fresh
+ // post-restore commits so any restore-time tag work has fully run
before we assert.
+ waitUntilCommittedFurther(table, 2);
+
+ // No savepoint tag must exist: S was aborted, so its tag would be
an orphan pointing at
+ // a savepoint that no longer exists.
+ assertThat(savepointTags(table)).isEmpty();
+ } finally {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ private FileStoreTable createTable(String tableName, boolean
coordinatorCommit)
+ throws Exception {
+ return createTable(tableName, coordinatorCommit, null);
+ }
+
+ private FileStoreTable createTable(
+ String tableName, boolean coordinatorCommit, String
commitCallbackClass)
+ throws Exception {
+ TableEnvironment tEnv =
+ TableEnvironment.create(
+
EnvironmentSettings.newInstance().inStreamingMode().build());
+ tEnv.executeSql(
+ "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' =
'"
+ + getTempDirPath()
+ + "' )");
+ tEnv.executeSql("USE CATALOG mycat");
+ // A stable operator-uid suffix so the writer's state maps back on
restore-from-savepoint.
+ String coordinatorOption =
+ coordinatorCommit
+ ? ", 'sink.coordinator-commit.enabled' = 'true',
'write-only' = 'true'"
+ : "";
+ String commitCallbackOption =
+ commitCallbackClass == null
+ ? ""
+ : ", 'commit.callbacks' = '" + commitCallbackClass +
"'";
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tableName
+ + " (id INT, data STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'sink.savepoint.auto-tag' = 'true', "
+ + "'commit.force-create-snapshot' = 'true', "
+ + "'sink.operator-uid.suffix' = 'failover-tag'"
+ + coordinatorOption
+ + commitCallbackOption
+ + ")");
+ return (FileStoreTable)
+ ((FlinkCatalog) tEnv.getCatalog("mycat").get())
+ .catalog()
+ .getTable(Identifier.create("default", tableName));
+ }
+
+ /** Runs the sink job at parallelism 1, optionally resuming from {@code
savepointPath}. */
+ private JobClient runSink(FileStoreTable table, String savepointPath)
throws Exception {
+ return runSink(table, savepointPath, false);
+ }
+
+ /**
+ * Runs the sink job at parallelism 1, optionally resuming from {@code
savepointPath}. When
+ * {@code blockCheckpointAfterSavepoint} is set, a {@link
BlockCheckpointAfterSavepointOperator}
+ * is chained in so that any normal checkpoint taken after a savepoint
stalls forever instead of
+ * completing — used to keep a savepoint's tag from being materialized by
a later checkpoint.
+ */
+ private JobClient runSink(
+ FileStoreTable table, String savepointPath, boolean
blockCheckpointAfterSavepoint)
+ throws Exception {
+ Configuration conf = new Configuration();
+ if (savepointPath != null) {
+ SavepointRestoreSettings.toConfiguration(
+ SavepointRestoreSettings.forPath(savepointPath, false),
conf);
+ }
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.setParallelism(1);
+ env.enableCheckpointing(200);
+ DataStreamSource<RowData> source =
+ env.fromSource(
+ new ContinuousSource(),
WatermarkStrategy.noWatermarks(), "restore-source");
+ DataStream<RowData> stream = source;
+ if (blockCheckpointAfterSavepoint) {
+ stream =
+ source.forward()
+ .transform(
+ "block-checkpoint-after-savepoint",
+ InternalTypeInfo.of(
+ RowType.of(new IntType(), new
VarCharType())),
+ new
BlockCheckpointAfterSavepointOperator());
+ }
+ new FlinkSinkBuilder(table).forRowData(stream).build();
+ return env.executeAsync("savepoint-failover-tag");
+ }
+
+ private void waitUntilSnapshotWithData(FileStoreTable table) throws
Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ while (System.currentTimeMillis() < deadline) {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ if (latest != null && latest.totalRecordCount() > 0) {
+ return;
+ }
+ Thread.sleep(200);
+ }
+ throw new IllegalStateException("no data-carrying snapshot committed
within timeout");
+ }
+
+ /** Waits until a snapshot committed after the current latest one, proving
the job resumed. */
+ private void waitUntilRecoveredAndCommitting(FileStoreTable table) throws
Exception {
+ Long baseline = table.snapshotManager().latestSnapshotId();
+ long base = baseline == null ? 0L : baseline;
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ while (System.currentTimeMillis() < deadline) {
+ Long latest = table.snapshotManager().latestSnapshotId();
+ if (latest != null && latest > base) {
+ return;
+ }
+ Thread.sleep(200);
+ }
+ throw new IllegalStateException("job did not resume committing after
region failover");
+ }
+
+ /**
+ * Waits until at least {@code count} more snapshots commit past the
current latest.
+ * Restore-time tag work runs before the resumed job commits again, so
requiring several fresh
+ * commits gives that work room to fully settle before we assert on tags.
+ */
+ private void waitUntilCommittedFurther(FileStoreTable table, int count)
throws Exception {
+ Long baseline = table.snapshotManager().latestSnapshotId();
+ long target = (baseline == null ? 0L : baseline) + count;
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ while (System.currentTimeMillis() < deadline) {
+ Long latest = table.snapshotManager().latestSnapshotId();
+ if (latest != null && latest >= target) {
+ return;
+ }
+ Thread.sleep(200);
+ }
+ throw new IllegalStateException("job did not commit enough snapshots
after restore");
+ }
+
+ private void waitUntilCallbackFired() throws Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ while (!FailOnFirstPostRecoveryCommitCallback.hasFired()
+ && System.currentTimeMillis() < deadline) {
+ Thread.sleep(200);
+ }
+ if (!FailOnFirstPostRecoveryCommitCallback.hasFired()) {
+ throw new IllegalStateException(
+ "commit callback never fired; the global failover was not
triggered");
+ }
+ }
+
+ private Map<Snapshot, List<String>>
waitUntilSavepointTagCreated(FileStoreTable table)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ Map<Snapshot, List<String>> tags = savepointTags(table);
+ while (tags.isEmpty() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(200);
+ tags = savepointTags(table);
+ }
+ assertThat(tags).describedAs("no savepoint tag was
created").isNotEmpty();
+ return tags;
+ }
+
+ private Map<Snapshot, List<String>> savepointTags(FileStoreTable table) {
+ return table.tagManager().tags(name ->
name.startsWith(SavepointTagUtils.PREFIX));
+ }
+
+ /** Emits one row per poll so every checkpoint window carries data. */
+ private static class ContinuousSource extends
AbstractNonCoordinatedSource<RowData> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<RowData, SimpleSourceSplit>
createReader(SourceReaderContext ctx) {
+ return new AbstractNonCoordinatedSourceReader<RowData>() {
+ private int next;
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<RowData> output)
+ throws InterruptedException {
+ output.collect(GenericRowData.of(next,
StringData.fromString("v" + next)));
+ next++;
+ Thread.sleep(20);
+ return InputStatus.MORE_AVAILABLE;
+ }
+ };
+ }
+ }
+
+ /**
+ * Passthrough operator that throws exactly once, on the target subtask,
while a savepoint is
+ * being taken. This reproduces "some writers finished snapshotState, one
did not", which forces
+ * a region failover with the savepoint in flight.
+ */
+ private static class FailOnSavepointOperator extends
AbstractStreamOperator<RowData>
+ implements OneInputStreamOperator<RowData, RowData> {
+ private static final long serialVersionUID = 1L;
+ private static final AtomicBoolean FAILED = new AtomicBoolean(false);
+ // Checkpoint id of the savepoint that was aborted, published by the
throwing subtask so the
+ // commit callback can tell "past the aborted savepoint" from ordinary
earlier commits.
+ private static volatile long savepointCheckpointId = -1L;
+ private final int targetSubtask;
+
+ FailOnSavepointOperator(int targetSubtask) {
+ this.targetSubtask = targetSubtask;
+ }
+
+ static void reset() {
+ FAILED.set(false);
+ savepointCheckpointId = -1L;
+ }
+
+ static long savepointCheckpointId() {
+ return savepointCheckpointId;
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) {
+ output.collect(element);
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation)
+ throws Exception {
+ int subtask =
getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+ if (checkpointOptions.getCheckpointType().isSavepoint()
+ && subtask == targetSubtask
+ && FAILED.compareAndSet(false, true)) {
+ savepointCheckpointId = checkpointId;
+ throw new RuntimeException(
+ "intentional region-failover trigger on subtask " +
subtask);
+ }
+ return super.snapshotState(checkpointId, timestamp,
checkpointOptions, storageLocation);
+ }
+ }
+
+ /**
+ * Passthrough operator that lets a savepoint through, but blocks every
normal checkpoint taken
+ * afterwards by parking in {@code snapshotState} forever. The blocked
checkpoint neither
+ * completes nor declines, so it cannot materialize a savepoint tag and —
unlike throwing — does
+ * not trip a failover; it only ends when the job is cancelled (which
interrupts this thread).
+ * Armed per test through a static flag so the topology-sharing sibling
case is unaffected.
+ */
+ private static class BlockCheckpointAfterSavepointOperator
+ extends AbstractStreamOperator<RowData>
+ implements OneInputStreamOperator<RowData, RowData> {
+ private static final long serialVersionUID = 1L;
+ private static volatile boolean armed = false;
+ private static volatile boolean savepointSeen = false;
+
+ static void armBlocking() {
+ armed = true;
+ savepointSeen = false;
+ }
+
+ static void disarmBlocking() {
+ armed = false;
+ savepointSeen = false;
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) {
+ output.collect(element);
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation)
+ throws Exception {
+ if (armed) {
+ if (checkpointOptions.getCheckpointType().isSavepoint()) {
+ savepointSeen = true;
+ } else if (savepointSeen) {
+ // Park until the job is cancelled; the interrupt ends the
wait.
+ synchronized (this) {
+ while (true) {
+ wait();
+ }
+ }
+ }
+ }
+ return super.snapshotState(checkpointId, timestamp,
checkpointOptions, storageLocation);
+ }
+ }
+
+ /**
+ * Commit callback loaded by {@code commit.callbacks} that throws exactly
once, on the first
+ * commit whose identifier is past the aborted savepoint's checkpoint id.
That commit is the
+ * first normal checkpoint after the region failover, so throwing there
forces a global failover
+ * whose latest completed checkpoint carries the surviving writer's stale
savepoint bit. Loaded
+ * reflectively by class name, so it must be public with a no-arg
constructor and coordinate
+ * through static fields (writer TM and JM coordinator share one JVM under
MiniCluster).
+ */
+ public static class FailOnFirstPostRecoveryCommitCallback implements
CommitCallback {
+ private static final AtomicBoolean FIRED = new AtomicBoolean(false);
+
+ public FailOnFirstPostRecoveryCommitCallback() {}
+
+ static void reset() {
+ FIRED.set(false);
+ }
+
+ static boolean hasFired() {
+ return FIRED.get();
+ }
+
+ @Override
+ public void call(Context context) {
+ long savepointId = FailOnSavepointOperator.savepointCheckpointId();
+ if (savepointId > 0
+ && context.identifier > savepointId
+ && FIRED.compareAndSet(false, true)) {
+ throw new RuntimeException(
+ "intentional global-failover trigger on commit " +
context.identifier);
+ }
+ }
+
+ @Override
+ public void retry(ManifestCommittable committable) {}
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
new file mode 100644
index 0000000000..759f5567bf
--- /dev/null
+++
b/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(
+ "Coordinator-commit stop-with-savepoint drops the async tag on
shutdownNow; re-enable"
+ + " after the coordinator end-input handling PR drains
pending tag work")
+ @ParameterizedTest(name = "coordinatorCommit = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(value = 180, unit = TimeUnit.SECONDS)
+ public void testStopWithSavepointCreatesTag(boolean coordinatorCommit)
throws Exception {
+ String tableName = coordinatorCommit ? "T_COORD_STOP" :
"T_CLASSIC_STOP";
+ FileStoreTable table = createTable(tableName, coordinatorCommit);
+
+ JobClient client = runSink(table);
+ // Wait until a data-carrying snapshot exists so the savepoint lands
on a checkpoint that
+ // carries data, avoiding the empty-savepoint boundary.
+ waitUntilSnapshotWithData(table);
+
+ // stop-with-savepoint (non-drain): the job terminates after the
savepoint, so there is no
+ // later checkpoint to fall back on — the tag must come from the
savepoint's own completion.
+ client.stopWithSavepoint(
+ false,
+ getTempDirPath("stop_savepoint_" + tableName),
+ SavepointFormatType.DEFAULT)
+ .get(120, TimeUnit.SECONDS);
+
+ 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();
+ }
+
+ private FileStoreTable createTable(String tableName, boolean
coordinatorCommit)
+ throws Exception {
+ TableEnvironment tEnv =
+ TableEnvironment.create(
+
EnvironmentSettings.newInstance().inStreamingMode().build());
+ tEnv.executeSql(
+ "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' =
'"
+ + getTempDirPath()
+ + "' )");
+ tEnv.executeSql("USE CATALOG mycat");
+ // force-create-snapshot ensures every completed checkpoint yields a
snapshot, keeping the
+ // DDL identical to the restore-tag parity IT.
+ String coordinatorOption =
+ coordinatorCommit
+ ? ", 'sink.coordinator-commit.enabled' = 'true',
'write-only' = 'true'"
+ : "";
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tableName
+ + " (id INT, data STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'sink.savepoint.auto-tag' = 'true', "
+ + "'commit.force-create-snapshot' = 'true'"
+ + coordinatorOption
+ + ")");
+ return (FileStoreTable)
+ ((FlinkCatalog) tEnv.getCatalog("mycat").get())
+ .catalog()
+ .getTable(Identifier.create("default", tableName));
+ }
+
+ private JobClient runSink(FileStoreTable table) throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(200);
+ DataStreamSource<RowData> stream =
+ env.fromSource(
+ new ContinuousSource(),
WatermarkStrategy.noWatermarks(), "tag-source");
+ new FlinkSinkBuilder(table).forRowData(stream).build();
+ return env.executeAsync("savepoint-tag");
+ }
+
+ private void waitUntilSnapshotWithData(FileStoreTable table) throws
Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ while (System.currentTimeMillis() < deadline) {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ if (latest != null && latest.totalRecordCount() > 0) {
+ return;
+ }
+ Thread.sleep(200);
+ }
+ throw new IllegalStateException("no data-carrying snapshot committed
within timeout");
+ }
+
+ private Map<Snapshot, List<String>>
waitUntilSavepointTagCreated(FileStoreTable table)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ Map<Snapshot, List<String>> tags = savepointTags(table);
+ while (tags.isEmpty() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(200);
+ tags = savepointTags(table);
+ }
+ assertThat(tags).describedAs("no savepoint tag was
created").isNotEmpty();
+ return tags;
+ }
+
+ private Map<Snapshot, List<String>> savepointTags(FileStoreTable table) {
+ return table.tagManager().tags(name ->
name.startsWith(SavepointTagUtils.PREFIX));
+ }
+
+ /** Emits one row per poll so every checkpoint window carries data. */
+ private static class ContinuousSource extends
AbstractNonCoordinatedSource<RowData> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<RowData, SimpleSourceSplit>
createReader(SourceReaderContext ctx) {
+ return new AbstractNonCoordinatedSourceReader<RowData>() {
+ private int next;
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<RowData> output)
+ throws InterruptedException {
+ output.collect(GenericRowData.of(next,
StringData.fromString("v" + next)));
+ next++;
+ Thread.sleep(20);
+ return InputStatus.MORE_AVAILABLE;
+ }
+ };
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
index bbd79573ec..1ebddec2bf 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperatorTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.utils.ThrowingConsumer;
@@ -91,8 +92,7 @@ public class AutoTagForSavepointCommitterOperatorTest extends
CommitterOperatorT
assertThat(snapshot.id()).isEqualTo(2);
Map<Snapshot, List<String>> tags = table.tagManager().tags();
assertThat(tags).containsOnlyKeys(snapshot);
- assertThat(tags.get(snapshot))
-
.containsOnly(AutoTagForSavepointCommitterOperator.SAVEPOINT_TAG_PREFIX + 2);
+
assertThat(tags.get(snapshot)).containsOnly(SavepointTagUtils.tagNameOf(2));
}
@Test
@@ -142,9 +142,7 @@ public class AutoTagForSavepointCommitterOperatorTest
extends CommitterOperatorT
Map<Snapshot, List<String>> tags = table.tagManager().tags();
assertThat(tags).containsOnlyKeys(snapshot);
- assertThat(tags.get(snapshot))
- .containsOnly(
-
AutoTagForSavepointCommitterOperator.SAVEPOINT_TAG_PREFIX + checkpointId);
+
assertThat(tags.get(snapshot)).containsOnly(SavepointTagUtils.tagNameOf(checkpointId));
}
@Test
@@ -181,6 +179,39 @@ public class AutoTagForSavepointCommitterOperatorTest
extends CommitterOperatorT
assertThat(table.tagManager().tagCount()).isEqualTo(0);
}
+ @Test
+ public void testAbortCheckpointKeepsTagFromDifferentCommitUser() throws
Exception {
+ FileStoreTable table = createFileStoreTable();
+ createSavepointTag(table, initialCommitUser + "-other", 1L);
+
+ OneInputStreamOperatorTestHarness<Committable, Committable>
testHarness =
+ createRecoverableTestHarness(table);
+ testHarness.open();
+ testHarness.getOneInputOperator().notifyCheckpointAborted(1L);
+ testHarness.close();
+
+
assertThat(table.tagManager().tagExists(SavepointTagUtils.tagNameOf(1L))).isTrue();
+ }
+
+ private void createSavepointTag(FileStoreTable table, String commitUser,
long commitIdentifier)
+ throws Exception {
+ try (StreamTableWrite write =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
+ StreamTableCommit commit =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newCommit()) {
+ write.write(GenericRow.of(1, 10L));
+ List<CommitMessage> messages = write.prepareCommit(false,
commitIdentifier);
+ commit.commit(commitIdentifier, messages);
+ }
+ table.tagManager()
+ .createTag(
+ table.snapshotManager().latestSnapshot(),
+ SavepointTagUtils.tagNameOf(commitIdentifier),
+ table.coreOptions().tagDefaultTimeRetained(),
+ table.store().createTagCallbacks(table),
+ false);
+ }
+
private void processCommittable(
OneInputStreamOperatorTestHarness<Committable, Committable>
testHarness,
StreamTableWrite write,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
index c533e175fa..b139aabe88 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
@@ -73,7 +73,7 @@ import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.fail;
/** Tests for {@link CommitterOperator}. */
-public class CommitterOperatorTest extends CommitterOperatorTestBase {
+public class CommitterOperatorTest extends CommitterTestBase {
protected String initialCommitUser;
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterTestBase.java
similarity index 96%
rename from
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
rename to
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterTestBase.java
index a69f8dbd3a..32f7b2b1da 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterTestBase.java
@@ -46,8 +46,10 @@ import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThat;
-/** Base test class for {@link CommitterOperatorTest}. */
-public abstract class CommitterOperatorTestBase {
+/**
+ * Base test class providing an unaware/fixed-bucket {@link FileStoreTable}
and result assertions.
+ */
+public abstract class CommitterTestBase {
private static final RowType ROW_TYPE =
RowType.of(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java
index 89911cc271..935f3bc99f 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java
@@ -34,9 +34,11 @@ import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.SavepointType;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.CoordinatorStore;
@@ -64,7 +66,7 @@ import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link CoordinatorCommittingRowDataStoreWriteOperator}. */
-public class CoordinatorCommittingRowDataStoreWriteOperatorTest extends
CommitterOperatorTestBase {
+public class CoordinatorCommittingRowDataStoreWriteOperatorTest extends
CommitterTestBase {
private static final TypeSerializer<CheckpointCommittables>
COMMITTABLES_SERIALIZER =
new SimpleVersionedSerializerTypeSerializerProxy<>(
@@ -91,7 +93,8 @@ public class
CoordinatorCommittingRowDataStoreWriteOperatorTest extends Committe
new StoreCommitter(
table,
table.newCommit(context.commitUser()), context),
true,
- commitUser);
+ commitUser,
+ null);
coordinator.start();
coordinator.waitProcessAllActions();
@@ -442,6 +445,121 @@ public class
CoordinatorCommittingRowDataStoreWriteOperatorTest extends Committe
secondHarness.close();
}
+ @Test
+ @Timeout(30)
+ public void testSavepointBitRidesOnCommittableEventAndPendingState()
throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ List<OperatorEvent> events = new ArrayList<>();
+
+ OneInputStreamOperatorTestHarness<InternalRow, Committable> harness =
+ createHarness(table, commitUser, events::add);
+ TypeSerializer<Committable> committableSerializer =
+ new CommittableTypeInfo().createSerializer(new
ExecutionConfig());
+ harness.setup(committableSerializer);
+ harness.open();
+ CoordinatorCommittingRowDataStoreWriteOperator operator =
+ (CoordinatorCommittingRowDataStoreWriteOperator)
harness.getOperator();
+
+ // cp1: a normal checkpoint carries savepoint=false in both the event
and the pending state.
+ harness.processElement(GenericRow.of(1, 10L), 1);
+ harness.prepareSnapshotPreBarrier(1);
+ harness.snapshot(1, 10);
+ assertThat(
+ ((CommittableEvent) events.get(0))
+ .deserialize(COMMITTABLES_SERIALIZER)
+ .shouldCreateSavepointTag())
+ .isFalse();
+
assertThat(operator.getPendingCommittables().get(1L).shouldCreateSavepointTag()).isFalse();
+
+ // cp2: a savepoint sets savepoint=true on both the emitted event and
the persisted entry.
+ harness.processElement(GenericRow.of(2, 20L), 2);
+ harness.prepareSnapshotPreBarrier(2);
+ harness.snapshotWithLocalState(
+ 2, 20, SavepointType.savepoint(SavepointFormatType.CANONICAL));
+ assertThat(
+ ((CommittableEvent) events.get(1))
+ .deserialize(COMMITTABLES_SERIALIZER)
+ .shouldCreateSavepointTag())
+ .isTrue();
+
assertThat(operator.getPendingCommittables().get(2L).shouldCreateSavepointTag()).isTrue();
+
+ harness.close();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testSavepointBitStaysFalseWhenAutoTagDisabled() throws
Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ List<OperatorEvent> events = new ArrayList<>();
+
+ OneInputStreamOperatorTestHarness<InternalRow, Committable> harness =
+ createHarness(table, commitUser, events::add, /*
autoTagForSavepoint */ false);
+ TypeSerializer<Committable> committableSerializer =
+ new CommittableTypeInfo().createSerializer(new
ExecutionConfig());
+ harness.setup(committableSerializer);
+ harness.open();
+ CoordinatorCommittingRowDataStoreWriteOperator operator =
+ (CoordinatorCommittingRowDataStoreWriteOperator)
harness.getOperator();
+
+ // Even on a savepoint, a writer without auto-tag enabled must not
flag a tag intent: the
+ // checkpoint is a savepoint, but no savepoint tag should be created
for it.
+ harness.processElement(GenericRow.of(1, 10L), 1);
+ harness.prepareSnapshotPreBarrier(1);
+ harness.snapshotWithLocalState(
+ 1, 10, SavepointType.savepoint(SavepointFormatType.CANONICAL));
+ assertThat(
+ ((CommittableEvent) events.get(0))
+ .deserialize(COMMITTABLES_SERIALIZER)
+ .shouldCreateSavepointTag())
+ .isFalse();
+
assertThat(operator.getPendingCommittables().get(1L).shouldCreateSavepointTag()).isFalse();
+
+ harness.close();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testSavepointBitReplayedOnRestore() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ TypeSerializer<Committable> committableSerializer =
+ new CommittableTypeInfo().createSerializer(new
ExecutionConfig());
+
+ // session 1: take a savepoint that is never notified complete, then
crash.
+ List<OperatorEvent> firstEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness<InternalRow, Committable>
firstHarness =
+ createHarness(table, commitUser, firstEvents::add);
+ firstHarness.setup(committableSerializer);
+ firstHarness.open();
+ firstHarness.processElement(GenericRow.of(1, 10L), 1);
+ firstHarness.prepareSnapshotPreBarrier(1);
+ OperatorSubtaskState snapshot =
+ firstHarness
+ .snapshotWithLocalState(
+ 1, 10,
SavepointType.savepoint(SavepointFormatType.CANONICAL))
+ .getJobManagerOwnedState();
+ firstHarness.close();
+
+ // session 2: restore replays the persisted savepoint bit in the
RestoredCommittableEvent.
+ List<OperatorEvent> restoredEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness<InternalRow, Committable>
secondHarness =
+ createHarness(table, commitUser, restoredEvents::add);
+ secondHarness.setup(committableSerializer);
+ restoreWithCheckpointId(secondHarness, snapshot, 1L);
+ secondHarness.open();
+
+ assertThat(restoredEvents).hasSize(1);
+ RestoredCommittableEvent restoredEvent = (RestoredCommittableEvent)
restoredEvents.get(0);
+ List<CheckpointCommittables> entries =
restoredEvent.deserialize(COMMITTABLES_SERIALIZER);
+ assertThat(entries).hasSize(1);
+ assertThat(entries.get(0).checkpointId()).isEqualTo(1L);
+ assertThat(entries.get(0).shouldCreateSavepointTag()).isTrue();
+
+ secondHarness.close();
+ }
+
private void assertCommittableEventCheckpoint(OperatorEvent event, long
expectedCheckpointId) {
CommittableEvent committableEvent = (CommittableEvent) event;
assertThat(committableEvent.getCheckpointId()).isEqualTo(expectedCheckpointId);
@@ -485,6 +603,15 @@ public class
CoordinatorCommittingRowDataStoreWriteOperatorTest extends Committe
private OneInputStreamOperatorTestHarness<InternalRow, Committable>
createHarness(
FileStoreTable table, String commitUser, OperatorEventGateway
gateway)
throws Exception {
+ return createHarness(table, commitUser, gateway, /*
autoTagForSavepoint */ true);
+ }
+
+ private OneInputStreamOperatorTestHarness<InternalRow, Committable>
createHarness(
+ FileStoreTable table,
+ String commitUser,
+ OperatorEventGateway gateway,
+ boolean autoTagForSavepoint)
+ throws Exception {
RowDataStoreWriteOperator.Factory operatorFactory =
new RowDataStoreWriteOperator.Factory(
table,
@@ -515,7 +642,8 @@ public class
CoordinatorCommittingRowDataStoreWriteOperatorTest extends Committe
table,
storeSinkWriteProvider,
commitUser,
- gateway);
+ gateway,
+ autoTagForSavepoint);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
index 91110d9962..0df52e25c4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
@@ -39,7 +39,7 @@ import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FlinkSink}. */
-public class FlinkSinkTest extends CommitterOperatorTestBase {
+public class FlinkSinkTest extends CommitterTestBase {
private static final RowType ROW_TYPE =
RowType.of(
@@ -105,17 +105,14 @@ public class FlinkSinkTest extends
CommitterOperatorTestBase {
}
@Test
- public void testCoordinatorCommitPreconditionsRejectsAutoTagForSavepoint()
throws Exception {
+ public void testCoordinatorCommitPreconditionsAllowsAutoTagForSavepoint()
throws Exception {
FileStoreTable table =
createUnawareBucketTable(
options ->
options.set(
FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT, true));
- assertThatThrownBy(
- () ->
- FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
- .isInstanceOf(IllegalArgumentException.class);
+ // auto-tag-for-savepoint is now supported on the coordinator-commit
path
+ FlinkSink.checkCoordinatorCommitPreconditions(table,
newCheckpointConfig(1), true);
}
@Test
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
new file mode 100644
index 0000000000..2e858e41c8
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/SavepointTagUtilsTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.tag.Tag;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+
+/** Tests for {@link SavepointTagUtils}. */
+public class SavepointTagUtilsTest extends CommitterTestBase {
+
+ @Test
+ public void testIsSavepointTagFor() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ Tag tag = createSavepointTag(table, "user", 1L, 1L);
+
+ assertThat(SavepointTagUtils.isSavepointTagFor(tag, "user",
1L)).isTrue();
+ assertThat(SavepointTagUtils.isSavepointTagFor(tag, "other-user",
1L)).isFalse();
+ assertThat(SavepointTagUtils.isSavepointTagFor(tag, "user",
2L)).isFalse();
+ }
+
+ @Test
+ public void testIsSavepointTagForWithNullCommitUser() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ Tag tag = createSavepointTag(table, "user", 1L, 1L);
+
+ assertThatNullPointerException()
+ .isThrownBy(() -> SavepointTagUtils.isSavepointTagFor(tag,
null, 1L));
+ }
+
+ @Test
+ public void testDeleteTagIfMatches() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ createSavepointTag(table, "user", 1L, 1L);
+
+ deleteTagIfMatches(table, "user", 1L);
+
+
assertThat(table.tagManager().tagExists(SavepointTagUtils.tagNameOf(1L))).isFalse();
+ }
+
+ @Test
+ public void testDeleteTagIfMatchesKeepsTagFromDifferentCommitUser() throws
Exception {
+ FileStoreTable table = createFileStoreTable();
+ createSavepointTag(table, "user", 1L, 1L);
+
+ deleteTagIfMatches(table, "other-user", 1L);
+
+
assertThat(table.tagManager().tagExists(SavepointTagUtils.tagNameOf(1L))).isTrue();
+ }
+
+ @Test
+ public void testDeleteTagIfMatchesKeepsTagForDifferentCommitIdentifier()
throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ createSavepointTag(table, "user", 1L, 2L);
+
+ deleteTagIfMatches(table, "user", 1L);
+
+
assertThat(table.tagManager().tagExists(SavepointTagUtils.tagNameOf(1L))).isTrue();
+ }
+
+ @Test
+ public void testDeleteTagIfMatchesWhenTagDoesNotExist() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+
+ deleteTagIfMatches(table, "user", 1L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ private Tag createSavepointTag(
+ FileStoreTable table,
+ String commitUser,
+ long savepointIdentifier,
+ long snapshotCommitIdentifier)
+ throws Exception {
+ try (StreamTableWrite write =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
+ StreamTableCommit commit =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newCommit()) {
+ write.write(GenericRow.of(1, 10L));
+ List<CommitMessage> messages = write.prepareCommit(false,
snapshotCommitIdentifier);
+ commit.commit(snapshotCommitIdentifier, messages);
+ }
+
+ String tagName = SavepointTagUtils.tagNameOf(savepointIdentifier);
+ table.tagManager()
+ .createTag(
+ table.snapshotManager().latestSnapshot(),
+ tagName,
+ table.coreOptions().tagDefaultTimeRetained(),
+ table.store().createTagCallbacks(table),
+ false);
+ return table.tagManager().get(tagName).get();
+ }
+
+ private void deleteTagIfMatches(
+ FileStoreTable table, String commitUser, long commitIdentifier) {
+ SavepointTagUtils.deleteTagIfMatches(
+ table.tagManager(),
+ commitUser,
+ commitIdentifier,
+ table.store().newTagDeletion(),
+ table.snapshotManager(),
+ table.store().createTagCallbacks(table));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
index 09b894d03d..e95ee7aef4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
@@ -38,8 +38,8 @@ public class CheckpointCommittablesSerializerTest {
new CommittableSerializer(new CommitMessageSerializer()));
@Test
- public void testCurrentVersionIsV2() {
- assertThat(serializer.getVersion()).isEqualTo(2);
+ public void testCurrentVersionIsV3() {
+ assertThat(serializer.getVersion()).isEqualTo(3);
}
@Test
@@ -53,10 +53,24 @@ public class CheckpointCommittablesSerializerTest {
assertThat(decoded.checkpointId()).isEqualTo(42L);
assertThat(decoded.watermark()).isEqualTo(4242L);
assertThat(decoded.idle()).isEqualTo(idle);
+ assertThat(decoded.shouldCreateSavepointTag()).isFalse();
assertThat(decoded.committables()).isEmpty();
}
}
+ @Test
+ public void testRoundTripPreservesSavepointFlag() throws IOException {
+ for (boolean savepoint : new boolean[] {true, false}) {
+ CheckpointCommittables original =
+ new CheckpointCommittables(
+ 42L, Collections.emptyList(), /* watermark */
4242L, false, savepoint);
+ CheckpointCommittables decoded =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(original));
+ assertThat(decoded.checkpointId()).isEqualTo(42L);
+
assertThat(decoded.shouldCreateSavepointTag()).isEqualTo(savepoint);
+ }
+ }
+
@Test
public void testV1PayloadDeserializesAsActive() throws IOException {
// Hand-encode a v1 payload (no idle bit) so the reader is exercised
on real bytes rather
@@ -73,6 +87,25 @@ public class CheckpointCommittablesSerializerTest {
// v1 predates idle tracking; readers must default to ACTIVE so
pre-upgrade payloads keep
// participating in the min just like they did before.
assertThat(decoded.idle()).isFalse();
+ assertThat(decoded.shouldCreateSavepointTag()).isFalse();
+ assertThat(decoded.committables()).isEmpty();
+ }
+
+ @Test
+ public void testV2PayloadDeserializesWithSavepointFalse() throws
IOException {
+ // Hand-encode a v2 payload (idle bit but no savepoint bit) to
exercise the v2->v3 fallback.
+ DataOutputSerializer out = new DataOutputSerializer(32);
+ out.writeLong(7L); // checkpointId
+ out.writeLong(1234L); // watermark
+ out.writeBoolean(true); // idle
+ out.writeInt(new CommittableSerializer(new
CommitMessageSerializer()).getVersion());
+ out.writeInt(0); // empty committables
+
+ CheckpointCommittables decoded = serializer.deserialize(2,
out.getCopyOfBuffer());
+ assertThat(decoded.checkpointId()).isEqualTo(7L);
+ assertThat(decoded.idle()).isTrue();
+ // v2 predates savepoint tracking; default to false (a normal
checkpoint).
+ assertThat(decoded.shouldCreateSavepointTag()).isFalse();
assertThat(decoded.committables()).isEmpty();
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
index 74f54da3aa..2357eaccc4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
@@ -27,7 +27,8 @@ import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.CommittableSerializer;
import org.apache.paimon.flink.sink.Committer;
-import org.apache.paimon.flink.sink.CommitterOperatorTestBase;
+import org.apache.paimon.flink.sink.CommitterTestBase;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
import org.apache.paimon.flink.sink.StoreCommitter;
import org.apache.paimon.flink.sink.state.CoordinatorState;
import org.apache.paimon.flink.sink.state.CoordinatorStateSerializer;
@@ -75,7 +76,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link CommittingWriteOperatorCoordinator}. */
-public class CommittingWriteOperatorCoordinatorTest extends
CommitterOperatorTestBase {
+public class CommittingWriteOperatorCoordinatorTest extends CommitterTestBase {
private static final TypeSerializer<CheckpointCommittables> SERIALIZER =
new SimpleVersionedSerializerTypeSerializerProxy<>(
@@ -376,6 +377,114 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
coordinator.close();
}
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testAutoTagForSavepointOnComplete() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 1);
+ CommittingWriteOperatorCoordinator coordinator =
+ createCoordinatorWithAutoTag(table, context);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ // savepoint at cp1 (not notified), then a normal cp2 completes and
materializes both.
+ coordinator.handleEventFromOperator(
+ 0, 0, savepointEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, 2,
2)));
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ Snapshot tagged = table.snapshotManager().snapshot(1);
+
assertThat(table.tagManager().tags().get(tagged)).containsOnly(savepointTag(1L));
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testAbortSavepointRemovesTag() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 1);
+ CommittingWriteOperatorCoordinator coordinator =
+ createCoordinatorWithAutoTag(table, context);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ // savepoint at cp1, then cp2 completes and creates the savepoint-1
tag.
+ coordinator.handleEventFromOperator(
+ 0, 0, savepointEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, 2,
2)));
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+
+ // aborting the savepoint removes the tag that a later checkpoint had
created.
+ coordinator.notifyCheckpointAborted(1L);
+ coordinator.waitProcessAllActions();
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testAutoTagForSavepointOnRestore() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 1);
+
+ // capture coordinator state holding an uncommitted savepoint at cp1.
+ CommittingWriteOperatorCoordinator first =
createCoordinatorWithAutoTag(table, context);
+ first.start();
+ first.handleEventFromOperator(
+ 0, 0, savepointEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ first.checkpointCoordinator(1, checkpoint);
+ first.waitProcessAllActions();
+ byte[] state = checkpoint.get();
+ first.close();
+ assertThat(table.latestSnapshot()).isNotPresent();
+
+ // restore: the replayed savepoint bit drives the re-commit and tag
creation while the
+ // coordinator keeps running — no intentional failover.
+ CommittingWriteOperatorCoordinator second =
createCoordinatorWithAutoTag(table, context);
+ second.resetToCheckpoint(1, state);
+ second.start();
+ second.waitProcessAllActions();
+ second.handleEventFromOperator(
+ 0,
+ 0,
+ savepointRestoreEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ second.waitProcessAllActions();
+
+ assertThat(failureCause).isNull();
+ assertThat(second.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING);
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ second.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testNoTagWhenAutoTagDisabled() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 1);
+ CommittingWriteOperatorCoordinator coordinator =
createCoordinator(table, context);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(
+ 0, 0, savepointEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ coordinator.close();
+ }
+
+ private static String savepointTag(long checkpointId) {
+ return SavepointTagUtils.tagNameOf(checkpointId);
+ }
+
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@Test
public void testCheckpointFutureCompletedExceptionallyOnSnapshotFailure()
throws Exception {
@@ -395,7 +504,8 @@ public class CommittingWriteOperatorCoordinatorTest extends
CommitterOperatorTes
commitContext),
expected),
true,
- commitUser);
+ commitUser,
+ null);
coordinator.start();
coordinator.waitProcessAllActions();
@@ -1087,7 +1197,31 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
.newCommit(),
commitContext),
true,
- commitUser);
+ commitUser,
+ null);
+ }
+
+ private CommittingWriteOperatorCoordinator createCoordinatorWithAutoTag(
+ FileStoreTable table, TestingContext context) {
+ return new CommittingWriteOperatorCoordinator(
+ context,
+ commitContext ->
+ new StoreCommitter(
+ table,
+ table.newStreamWriteBuilder()
+
.withCommitUser(commitContext.commitUser())
+ .newCommit(),
+ commitContext),
+ true,
+ commitUser,
+ user ->
+ new SavepointTagger(
+ table.snapshotManager(),
+ table.tagManager(),
+ table.store().newTagDeletion(),
+ table.store().createTagCallbacks(table),
+ table.coreOptions().tagDefaultTimeRetained(),
+ user));
}
private CommittingWriteOperatorCoordinator
createCoordinatorCapturingContext(
@@ -1106,7 +1240,8 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
commitContext);
},
true,
- commitUser);
+ commitUser,
+ null);
}
private Committable committable(FileStoreTable table, long checkpointId,
int value)
@@ -1182,6 +1317,34 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
return eventOf(checkpointId, Collections.emptyList(), Long.MIN_VALUE);
}
+ private CommittableEvent savepointEvent(long checkpointId,
List<Committable> committables)
+ throws Exception {
+ return CommittableEvent.create(
+ checkpointId,
+ new CheckpointCommittables(
+ checkpointId,
+ committables,
+ Long.MIN_VALUE,
+ /* idle */ false,
+ /* shouldCreateSavepointTag */ true),
+ SERIALIZER);
+ }
+
+ private RestoredCommittableEvent savepointRestoreEvent(
+ long restoredCheckpointId, List<Committable> committables) throws
Exception {
+ CheckpointCommittables checkpointCommittables =
+ new CheckpointCommittables(
+ restoredCheckpointId,
+ committables,
+ Long.MIN_VALUE,
+ /* idle */ false,
+ /* shouldCreateSavepointTag */ true);
+ return RestoredCommittableEvent.create(
+ restoredCheckpointId,
+ Collections.singletonList(checkpointCommittables),
+ SERIALIZER);
+ }
+
private RestoredCommittableEvent restoreEvent(
long restoredCheckpointId, Committable committable) throws
Exception {
return restoreEventOf(restoredCheckpointId,
Collections.singletonList(committable));
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
new file mode 100644
index 0000000000..a84bc205c9
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SavepointTaggerTest.java
@@ -0,0 +1,243 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.sink.CommitterTestBase;
+import org.apache.paimon.flink.sink.SavepointTagUtils;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link SavepointTagger}. */
+public class SavepointTaggerTest extends CommitterTestBase {
+
+ private String commitUser;
+
+ @BeforeEach
+ public void before() {
+ super.before();
+ commitUser = UUID.randomUUID().toString();
+ }
+
+ @Test
+ public void testTagUpToCreatesTagForCommittedSavepoint() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(1L);
+ commitSnapshot(table, 1L);
+ commitSnapshot(table, 2L);
+
+ tagger.tagUpTo(2L);
+
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+ }
+
+ @Test
+ public void testTagUpToBoundaryIsInclusive() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(2L);
+ commitSnapshot(table, 2L);
+
+ // tagUpTo uses headSet(checkpointId, true), so a pending id equal to
checkpointId is
+ // tagged.
+ tagger.tagUpTo(2L);
+
+ assertThat(table.tagManager().tagExists(savepointTag(2L))).isTrue();
+ }
+
+ @Test
+ public void testTagUpToLeavesLaterPendingUntagged() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(5L);
+ commitSnapshot(table, 5L);
+
+ // 5 is above the checkpoint watermark, so it stays pending and no tag
is created.
+ tagger.tagUpTo(4L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testTagUpToWithNoPendingCreatesNothing() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ commitSnapshot(table, 1L);
+
+ tagger.tagUpTo(1L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testTagUpToIsIdempotent() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(1L);
+ commitSnapshot(table, 1L);
+ commitSnapshot(table, 2L);
+
+ // A later checkpoint may re-tag an already-tagged snapshot; the
second call is a no-op.
+ tagger.tagUpTo(2L);
+ tagger.tagUpTo(2L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+ }
+
+ @Test
+ public void testDropAbortedRemovesCreatedTag() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(1L);
+ commitSnapshot(table, 1L);
+ commitSnapshot(table, 2L);
+ tagger.tagUpTo(2L);
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+
+ // A cumulative commit may have tagged an aborted savepoint; dropping
it removes the tag.
+ tagger.dropAborted(1L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testDropAbortedWithoutTagIsNoop() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ commitSnapshot(table, 1L);
+
+ tagger.dropAborted(1L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testDropAbortedRemovesPendingBeforeTagging() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ tagger.add(1L);
+ // Aborted before any tagging round, so the pending intent must not
survive to tagUpTo.
+ tagger.dropAborted(1L);
+ commitSnapshot(table, 1L);
+ commitSnapshot(table, 2L);
+
+ tagger.tagUpTo(2L);
+
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testDropAbortedKeepsTagFromDifferentCommitUser() throws
Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ createSavepointTag(table, UUID.randomUUID().toString(), 1L, 1L);
+
+ tagger.dropAborted(1L);
+
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ }
+
+ @Test
+ public void testDropAbortedKeepsTagForDifferentCommitIdentifier() throws
Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ SavepointTagger tagger = createTagger(table);
+
+ createSavepointTag(table, commitUser, 1L, 2L);
+
+ tagger.dropAborted(1L);
+
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ }
+
+ private SavepointTagger createTagger(FileStoreTable table) {
+ return new SavepointTagger(
+ table.snapshotManager(),
+ table.tagManager(),
+ table.store().newTagDeletion(),
+ table.store().createTagCallbacks(table),
+ table.coreOptions().tagDefaultTimeRetained(),
+ commitUser);
+ }
+
+ private void commitSnapshot(FileStoreTable table, long commitIdentifier)
throws Exception {
+ commitSnapshot(table, commitUser, commitIdentifier);
+ }
+
+ private void createSavepointTag(
+ FileStoreTable table,
+ String commitUser,
+ long commitIdentifier,
+ long snapshotCommitIdentifier)
+ throws Exception {
+ commitSnapshot(table, commitUser, snapshotCommitIdentifier);
+ table.tagManager()
+ .createTag(
+ table.snapshotManager().latestSnapshot(),
+ savepointTag(commitIdentifier),
+ table.coreOptions().tagDefaultTimeRetained(),
+ table.store().createTagCallbacks(table),
+ false);
+ }
+
+ private void commitSnapshot(FileStoreTable table, String commitUser, long
commitIdentifier)
+ throws Exception {
+ try (StreamTableWrite write =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
+ StreamTableCommit commit =
+
table.newStreamWriteBuilder().withCommitUser(commitUser).newCommit()) {
+ write.write(GenericRow.of((int) commitIdentifier,
commitIdentifier));
+ List<CommitMessage> messages = write.prepareCommit(false,
commitIdentifier);
+ commit.commit(commitIdentifier, messages);
+ }
+ }
+
+ private FileStoreTable createUnawareBucketTable() throws Exception {
+ return createFileStoreTable(
+ options -> {
+ options.set(CoreOptions.BUCKET, -1);
+ options.remove("bucket-key");
+ });
+ }
+
+ private static String savepointTag(long checkpointId) {
+ return SavepointTagUtils.tagNameOf(checkpointId);
+ }
+}
