diff --git a/docs/generated/flink_connector_configuration.html b/docs/generated/flink_connector_configuration.html
index 8f5b2eb9c3bb..1d0b68bb94e4 100644
--- a/docs/generated/flink_connector_configuration.html
+++ b/docs/generated/flink_connector_configuration.html
@@ -276,7 +276,7 @@
sink.coordinator-commit.enabled |
false |
Boolean |
- If true, run the Paimon committer inside the Flink JobManager via an OperatorCoordinator. This decouples commit from any single TaskManager subtask so that region failover does not have to restart the whole pipeline. Only supports unaware-bucket append tables in streaming mode with checkpointing enabled; unsupported configurations fail during sink planning. |
+ If true, run the Paimon committer inside the Flink JobManager via an OperatorCoordinator. This decouples commit from any single TaskManager subtask so that region failover does not have to restart the whole pipeline. Only supports unaware-bucket append tables. Streaming mode requires checkpointing enabled; batch mode commits after all writers reach end input. Unsupported configurations fail during sink planning. |
sink.cross-partition.managed-memory |
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index 1f0bf32c0956..851c3d280dd8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -504,8 +504,9 @@ public class FlinkConnectorOptions {
"If true, run the Paimon committer inside the Flink JobManager via an "
+ "OperatorCoordinator. This decouples commit from any single TaskManager "
+ "subtask so that region failover does not have to restart the whole pipeline. "
- + "Only supports unaware-bucket append tables in streaming mode with "
- + "checkpointing enabled; unsupported configurations fail during sink planning.");
+ + "Only supports unaware-bucket append tables. Streaming mode requires "
+ + "checkpointing enabled; batch mode commits after all writers reach end input. "
+ + "Unsupported configurations fail during sink planning.");
public static final ConfigOption SINK_WRITER_COORDINATOR_ENABLED =
key("sink.writer-coordinator.enabled")
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 3bd617b5eb5b..7b55aba6de03 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
@@ -65,6 +65,7 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
private static final Logger LOG =
LoggerFactory.getLogger(CoordinatorCommittingRowDataStoreWriteOperator.class);
+ private static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
@VisibleForTesting
static final String PENDING_COMMITTABLE_STATE_NAME = "pending_committable_state";
@@ -134,6 +135,11 @@ public void initializeState(StateInitializationContext context) throws Exception
List restored = new ArrayList<>();
for (CheckpointCommittables entry : pendingCommittableState.get()) {
+ // End input is newer than every ordinary restored checkpoint and must survive
+ // subsequent snapshots even if Flink does not call endInput again after restore.
+ if (entry.checkpointId() == END_INPUT_CHECKPOINT_ID) {
+ pendingCommittables.put(entry.checkpointId(), entry);
+ }
restored.add(entry);
}
pendingCommittableState.clear();
@@ -172,6 +178,18 @@ protected void emitCommittables(boolean waitCompaction, long checkpointId) throw
CheckpointCommittables entry =
new CheckpointCommittables(
checkpointId, committables, currentWatermark, currentIdle);
+ if (checkpointId == END_INPUT_CHECKPOINT_ID) {
+ CheckpointCommittables previous = pendingCommittables.get(checkpointId);
+ if (previous != null) {
+ // A restored writer may receive endInput again. Preserve the previously persisted
+ // final committables and send one authoritative entry to the coordinator.
+ List merged = new ArrayList<>(previous.committables());
+ merged.addAll(entry.committables());
+ entry =
+ new CheckpointCommittables(
+ checkpointId, merged, currentWatermark, currentIdle);
+ }
+ }
// Emit an event per (subtask, checkpoint) regardless of whether committables is empty.
operatorEventGateway.sendEventToCoordinator(
CommittableEvent.create(checkpointId, entry, eventSerializer));
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 b323301c7933..ccd72a8c9169 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
@@ -53,6 +53,7 @@
import java.util.Queue;
import java.util.Set;
+import static org.apache.paimon.CoreOptions.TAG_AUTOMATIC_CREATION;
import static org.apache.paimon.CoreOptions.WRITE_ONLY;
import static org.apache.paimon.CoreOptions.createCommitUser;
import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK;
@@ -135,6 +136,9 @@ public DataStream doWrite(
DataStream input, String commitUser, @Nullable Integer parallelism) {
StreamExecutionEnvironment env = input.getExecutionEnvironment();
boolean isStreaming = isStreaming(input);
+ boolean streamingCheckpointEnabled =
+ isStreaming && env.getCheckpointConfig().isCheckpointingEnabled();
+ Options options = Options.fromMap(table.options());
boolean writeOnly = table.coreOptions().writeOnly();
StoreSinkWrite.Provider writeProvider =
@@ -152,15 +156,17 @@ public DataStream doWrite(
input.transform(
(writeOnly ? WRITER_WRITE_ONLY_NAME : WRITER_NAME) + " : " + table.name(),
new CommittableTypeInfo(),
- createWriteOperatorFactory(writeProvider, commitUser));
+ createWriteOperatorFactory(
+ writeProvider,
+ commitUser,
+ streamingCheckpointEnabled,
+ options.get(END_INPUT_WATERMARK)));
if (parallelism == null) {
forwardParallelism(written, input);
} else {
written.setParallelism(parallelism);
}
- Options options = Options.fromMap(table.options());
-
String uidSuffix = options.get(SINK_OPERATOR_UID_SUFFIX);
if (options.get(SINK_OPERATOR_UID_SUFFIX) != null) {
written = written.uid(generateCustomUid(WRITER_NAME, table.name(), uidSuffix));
@@ -202,14 +208,16 @@ public DataStream doWrite(
public DataStreamSink doCommit(DataStream written, String commitUser) {
StreamExecutionEnvironment env = written.getExecutionEnvironment();
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
+ boolean isStreaming = isStreaming(written);
boolean streamingCheckpointEnabled =
- isStreaming(written) && checkpointConfig.isCheckpointingEnabled();
+ isStreaming && checkpointConfig.isCheckpointingEnabled();
if (streamingCheckpointEnabled) {
assertStreamingConfiguration(env);
}
if (coordinatorCommitEnabled()) {
- return doCoordinatorCommit(written, checkpointConfig, streamingCheckpointEnabled);
+ return doCoordinatorCommit(
+ written, checkpointConfig, isStreaming, streamingCheckpointEnabled);
}
return doOperatorCommit(written, commitUser, streamingCheckpointEnabled);
}
@@ -217,8 +225,10 @@ public DataStreamSink doCommit(DataStream written, String commit
private DataStreamSink doCoordinatorCommit(
DataStream written,
CheckpointConfig checkpointConfig,
+ boolean isStreaming,
boolean streamingCheckpointEnabled) {
- checkCoordinatorCommitPreconditions(table, checkpointConfig, streamingCheckpointEnabled);
+ checkCoordinatorCommitPreconditions(
+ table, checkpointConfig, isStreaming, streamingCheckpointEnabled);
// The commit runs inside the writer's OperatorCoordinator on the JobManager, so there
// is no global committer operator. Committables are still forwarded by the writer for
// observability and are discarded here.
@@ -341,6 +351,14 @@ public static void assertBatchAdaptiveParallelism(
endInputWatermark);
}
+ protected OneInputStreamOperatorFactory createWriteOperatorFactory(
+ StoreSinkWrite.Provider writeProvider,
+ String commitUser,
+ boolean streamingCheckpointEnabled,
+ @Nullable Long endInputWatermark) {
+ return createWriteOperatorFactory(writeProvider, commitUser);
+ }
+
protected abstract OneInputStreamOperatorFactory createWriteOperatorFactory(
StoreSinkWrite.Provider writeProvider, String commitUser);
@@ -368,15 +386,15 @@ protected boolean coordinatorCommitEnabled() {
static void checkCoordinatorCommitPreconditions(
FileStoreTable table,
CheckpointConfig checkpointConfig,
+ boolean isStreaming,
boolean streamingCheckpointEnabled) {
Options options = Options.fromMap(table.options());
- // Region failover only benefits streaming jobs. A batch job persists its shuffle data and
- // has no region-failover concern, so coordinator commit brings no benefit and batch is not
- // supported. The commit is driven by checkpoint completion, so checkpointing must be on.
+ // Streaming commits are driven by checkpoint completion. Batch jobs commit when every
+ // writer reaches endInput, so they do not require checkpointing.
checkArgument(
- streamingCheckpointEnabled,
- "Could not enable coordinator commit because it requires streaming mode with checkpointing enabled.");
+ !isStreaming || streamingCheckpointEnabled,
+ "Could not enable coordinator commit in streaming mode without checkpointing.");
// The checks below reject configurations that introduce an all-to-all shuffle. Such a
// shuffle places every operator into a single failover region, which defeats the region
@@ -407,6 +425,12 @@ static void checkCoordinatorCommitPreconditions(
"Could not enable coordinator commit because "
+ SINK_AUTO_TAG_FOR_SAVEPOINT.key()
+ " is enabled, which is not supported yet.");
+ // TODO support batch tag create.
+ checkArgument(
+ table.coreOptions().tagCreationMode() != TagCreationMode.BATCH,
+ "Could not enable coordinator commit because "
+ + TAG_AUTOMATIC_CREATION.key()
+ + " = batch is not supported yet.");
// TODO concurrent checkpoints are not supported yet.
checkArgument(
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 b926dae4c2e5..fd41255e1292 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
@@ -33,6 +33,8 @@
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import javax.annotation.Nullable;
+
import java.util.Map;
/** An {@link AppendTableSink} which handles {@link InternalRow}. */
@@ -48,16 +50,24 @@ public RowAppendTableSink(
@Override
protected OneInputStreamOperatorFactory createWriteOperatorFactory(
StoreSinkWrite.Provider writeProvider, String commitUser) {
+ return createWriteOperatorFactory(writeProvider, commitUser, true, null);
+ }
+
+ @Override
+ protected OneInputStreamOperatorFactory createWriteOperatorFactory(
+ StoreSinkWrite.Provider writeProvider,
+ String commitUser,
+ boolean streamingCheckpointEnabled,
+ @Nullable Long endInputWatermark) {
if (coordinatorCommitEnabled()) {
return createCoordinatorCommittingRowWriteOperatorFactory(
table,
writeProvider,
commitUser,
- // checkpointing on by default for the JM-side committer; bounded sources will
- // be handled by end-input support in a follow-up PR
- true,
+ streamingCheckpointEnabled,
true,
- createCommitterFactory());
+ createCommitterFactory(),
+ endInputWatermark);
}
return createNoStateRowWriteOperatorFactory(table, writeProvider, commitUser);
}
@@ -84,14 +94,16 @@ protected CommittableStateManager createCommittableStateMan
String commitUser,
boolean streamingCheckpointEnabled,
boolean failoverAfterRecovery,
- Committer.Factory committerFactory) {
+ Committer.Factory committerFactory,
+ @Nullable Long endInputWatermark) {
return new CoordinatorCommittingFactory(
table,
writeProvider,
commitUser,
streamingCheckpointEnabled,
failoverAfterRecovery,
- committerFactory);
+ committerFactory,
+ endInputWatermark);
}
private static class CoordinatorCommittingFactory extends RowDataStoreWriteOperator.Factory
@@ -102,6 +114,7 @@ private static class CoordinatorCommittingFactory extends RowDataStoreWriteOpera
private final boolean streamingCheckpointEnabled;
private final boolean failoverAfterRecovery;
private final Committer.Factory committerFactory;
+ private final Long endInputWatermark;
CoordinatorCommittingFactory(
FileStoreTable table,
@@ -109,11 +122,13 @@ private static class CoordinatorCommittingFactory extends RowDataStoreWriteOpera
String initialCommitUser,
boolean streamingCheckpointEnabled,
boolean failoverAfterRecovery,
- Committer.Factory committerFactory) {
+ Committer.Factory committerFactory,
+ @Nullable Long endInputWatermark) {
super(table, storeSinkWriteProvider, initialCommitUser);
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.failoverAfterRecovery = failoverAfterRecovery;
this.committerFactory = committerFactory;
+ this.endInputWatermark = endInputWatermark;
}
@Override
@@ -124,7 +139,8 @@ public OperatorCoordinator.Provider getCoordinatorProvider(
committerFactory,
streamingCheckpointEnabled,
initialCommitUser,
- failoverAfterRecovery);
+ failoverAfterRecovery,
+ endInputWatermark);
}
@Override
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 20f3b65b8865..53dcfde6f485 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.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -50,6 +52,7 @@
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -72,20 +75,21 @@ public class CommittingWriteOperatorCoordinator implements OperatorCoordinator {
private static final Logger LOG =
LoggerFactory.getLogger(CommittingWriteOperatorCoordinator.class);
+ private static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
private final OperatorCoordinator.Context context;
private final Committer.Factory committerFactory;
private final boolean streamingCheckpointEnabled;
private final boolean failoverAfterRecovery;
- private final int parallelism;
- private final WriterCommittables[] subtaskCommittables;
+ private WriterCommittables[] subtaskCommittables;
private final TypeSerializer committablesSerializer;
private final CoordinatorStateSerializer stateSerializer;
- private final ExecutorService commitExecutor;
+ private CoordinatorExecutorThreadFactory coordinatorThreadFactory;
+ private ExecutorService commitExecutor;
// Rebuilt per coordinator instance; state is purely in-memory, matching Flink's
// StatusWatermarkValve which is also reconstructed per task instance without checkpointing.
- private final WatermarkAligner watermarkAligner;
+ private WatermarkAligner watermarkAligner;
// Populated by resetToCheckpoint and consumed by start. Plain fields are sufficient: both
// callbacks run on the same scheduler thread in order.
@@ -96,31 +100,30 @@ public class CommittingWriteOperatorCoordinator implements OperatorCoordinator {
private Committer committer;
private String commitUser;
private MemoryBackendStateStore stateStore;
+ private boolean endInputCommitted;
+ @Nullable private final Long endInputWatermark;
public CommittingWriteOperatorCoordinator(
OperatorCoordinator.Context context,
Committer.Factory committerFactory,
boolean streamingCheckpointEnabled,
String initialCommitUser,
- boolean failoverAfterRecovery) {
+ boolean failoverAfterRecovery,
+ @Nullable Long endInputWatermark) {
this.context = context;
this.committerFactory = committerFactory;
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.commitUser = initialCommitUser;
this.failoverAfterRecovery = failoverAfterRecovery;
- this.parallelism = context.currentParallelism();
- this.subtaskCommittables = new WriterCommittables[parallelism];
+ this.endInputWatermark = endInputWatermark;
this.committablesSerializer =
new SimpleVersionedSerializerTypeSerializerProxy<>(
() ->
new CheckpointCommittablesSerializer(
new CommittableSerializer(new CommitMessageSerializer())));
this.stateSerializer = new CoordinatorStateSerializer();
- this.commitExecutor =
- Executors.newSingleThreadExecutor(
- new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context));
- this.watermarkAligner = new WatermarkAligner(parallelism);
this.state = State.CREATED;
+ this.endInputCommitted = false;
}
@Override
@@ -131,6 +134,15 @@ public void start() throws Exception {
state == State.CREATED || state == State.RESTORING,
"Coordinator already started, illegal state %s",
state);
+ // RecreateOnResetOperatorCoordinator constructs the inner coordinator with a lazily
+ // initialized Context. In particular, AdaptiveBatchScheduler creates the inner instance
+ // before currentParallelism() is available. start() is the first lifecycle callback where
+ // the Context is guaranteed to be initialized.
+ subtaskCommittables = new WriterCommittables[context.currentParallelism()];
+ this.watermarkAligner = new WatermarkAligner(context.currentParallelism());
+ coordinatorThreadFactory =
+ new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context);
+ commitExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory);
runInEventLoop(
() -> {
if (state == State.RESTORING) {
@@ -150,6 +162,9 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
+ if (commitExecutor != null) {
+ drainCommitEventLoop();
+ }
transitionState(State.CLOSED);
if (commitExecutor != null) {
commitExecutor.shutdownNow();
@@ -194,7 +209,6 @@ public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEven
} else if (event instanceof RestoredCommittableEvent) {
handleRestoredCommittableEvent(subtask, (RestoredCommittableEvent) event);
} else {
- // TODO: end input handling
throw new UnsupportedOperationException("Unsupported event type: " + event);
}
},
@@ -204,6 +218,25 @@ public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEven
attemptNumber);
}
+ /** Waits until all actions submitted before coordinator close have completed. */
+ private void drainCommitEventLoop() {
+ checkState(
+ coordinatorThreadFactory == null
+ || !coordinatorThreadFactory.isCurrentThreadCoordinatorThread(),
+ "Cannot drain the commit event loop from its own thread.");
+ CompletableFuture completion = new CompletableFuture<>();
+ runInEventLoop(() -> completion.complete(null), "draining commit event loop before close");
+
+ try {
+ completion.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for the commit event loop", e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Failed to drain the commit event loop", e.getCause());
+ }
+ }
+
@Override
public void notifyCheckpointComplete(long checkpointId) {
runInEventLoop(
@@ -213,29 +246,45 @@ public void notifyCheckpointComplete(long checkpointId) {
"Completing checkpoint should be notified in RUNNING state, while current state is "
+ state);
}
+ if (endInputCommitted) {
+ LOG.debug(
+ "Ignore completion of checkpoint {} because end input has already been committed.",
+ checkpointId);
+ return;
+ }
// writers always report a committable per (subtask, checkpoint) during
- // snapshot, even if empty; missing means the writer is broken
+ // snapshot until they finish. An ended writer covers every later ordinary
+ // checkpoint because it cannot produce more data.
if (!alignCommittables(checkpointId)) {
throw new IllegalStateException("Not all committables reported by writer");
}
+ boolean finalCommit = allSubtasksEndInput();
+ long targetCheckpointId = finalCommit ? END_INPUT_CHECKPOINT_ID : checkpointId;
Map watermarkPerCheckpoint =
- alignWatermarkPerCheckpoint(
- checkpointId, subtaskCommittables, watermarkAligner);
+ alignWatermarkPerCheckpointForCommit(
+ targetCheckpointId, subtaskCommittables, watermarkAligner);
commitUpToCheckpoint(
- checkpointId,
+ targetCheckpointId,
pollManifestCommittablesForCheckpoint(
- checkpointId,
+ targetCheckpointId,
subtaskCommittables,
watermarkPerCheckpoint,
committer),
watermarkPerCheckpoint,
committables -> {
try {
- committer.commit(committables);
+ if (finalCommit) {
+ committer.filterAndCommit(committables, false, true);
+ } else {
+ committer.commit(committables);
+ }
} catch (Exception e) {
throw new RuntimeException(e);
}
});
+ if (finalCommit) {
+ endInputCommitted = true;
+ }
},
"completing checkpoint %d",
checkpointId);
@@ -302,8 +351,15 @@ public void executionAttemptReady(int subtask, int attemptNumber, SubtaskGateway
private void handleCommittableEvent(int subtask, CommittableEvent event) throws Exception {
if (state == State.RUNNING) {
+ if (endInputCommitted && event.getCheckpointId() == END_INPUT_CHECKPOINT_ID) {
+ LOG.debug(
+ "Ignore repeated end input event from subtask {} after final commit.",
+ subtask);
+ return;
+ }
updateSubtaskCommittables(
subtask, WriterCommittables.from(event, committablesSerializer));
+ commitEndInputIfCheckpointDisabled();
} else {
throw new IllegalStateException(
"Illegal state " + state + " while handling committable event " + event);
@@ -316,16 +372,30 @@ private void handleRestoredCommittableEvent(int subtask, RestoredCommittableEven
updateSubtaskCommittables(
subtask, WriterCommittables.fromRestore(event, committablesSerializer));
if (alignCommittables(event.getRestoredCheckpointId())) {
- recover(event.getRestoredCheckpointId());
+ boolean finalCommit = allSubtasksEndInput();
+ recover(finalCommit ? END_INPUT_CHECKPOINT_ID : event.getRestoredCheckpointId());
+ if (finalCommit) {
+ endInputCommitted = true;
+ }
transitionState(State.RUNNING);
}
} else if (state == State.RUNNING) {
- // a region failover replayed restore committables while the coordinator itself is
- // not restoring; it already holds the committed state, so ignore them
- LOG.info(
- "Ignore restore committables from subtask {} of checkpoint {}, coordinator is running.",
- subtask,
- event.getRestoredCheckpointId());
+ // A region failover replays entries from an already committed checkpoint. Ordinary
+ // entries can be ignored, but an end-input entry is newer than every ordinary
+ // checkpoint and may still be waiting for the other subtasks to finish.
+ WriterCommittables restored =
+ WriterCommittables.fromRestore(event, committablesSerializer);
+ if (restored.isEndInput() && !endInputCommitted) {
+ // region failover will reset
+ updateSubtaskCommittables(
+ subtask, new WriterCommittables(restored.getEndInputCommittables()));
+ commitEndInputIfCheckpointDisabled();
+ } else {
+ LOG.info(
+ "Ignore restore committables from subtask {} of checkpoint {}, coordinator is running.",
+ subtask,
+ event.getRestoredCheckpointId());
+ }
} else {
throw new IllegalStateException(
"Illegal state "
@@ -345,19 +415,48 @@ private void updateSubtaskCommittables(int subtask, WriterCommittables incoming)
private boolean alignCommittables(long checkpointId) {
for (WriterCommittables committables : subtaskCommittables) {
- if (committables == null || committables.getMaxCheckpointId() < checkpointId) {
+ if (committables == null || !committables.coversCheckpoint(checkpointId)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private boolean allSubtasksEndInput() {
+ for (WriterCommittables committables : subtaskCommittables) {
+ if (committables == null || !committables.isEndInput()) {
return false;
}
}
return true;
}
+ private void commitEndInputIfCheckpointDisabled() throws Exception {
+ if (streamingCheckpointEnabled || endInputCommitted || !allSubtasksEndInput()) {
+ return;
+ }
+
+ Map watermarkPerCheckpoint =
+ alignWatermarkPerCheckpointForCommit(
+ END_INPUT_CHECKPOINT_ID, subtaskCommittables, watermarkAligner);
+ commitUpToCheckpoint(
+ END_INPUT_CHECKPOINT_ID,
+ pollManifestCommittablesForCheckpoint(
+ END_INPUT_CHECKPOINT_ID,
+ subtaskCommittables,
+ watermarkPerCheckpoint,
+ committer),
+ watermarkPerCheckpoint,
+ committables -> committer.filterAndCommit(committables, false, true));
+ endInputCommitted = true;
+ }
+
// replaces CommittableStateManager because committables are not stored in the committer
private void recover(long checkpointId) throws Exception {
if (failoverAfterRecovery) {
// recommit the restored committables and trigger a failover to reinitialize all writers
Map watermarkPerCheckpoint =
- alignWatermarkPerCheckpoint(
+ alignWatermarkPerCheckpointForCommit(
checkpointId, subtaskCommittables, watermarkAligner);
commitUpToCheckpoint(
checkpointId,
@@ -456,6 +555,20 @@ private static SubtaskWatermark[] subtaskWatermarksAt(
return subtaskWatermarks;
}
+ /**
+ * Reduces the per-subtask watermarks into the specified {@code endInputWatermark} within the
+ * committable when EndInput is encountered.
+ */
+ private Map alignWatermarkPerCheckpointForCommit(
+ long checkpointId, WriterCommittables[] subtaskCommittables, WatermarkAligner aligner) {
+ Map watermarkPerCheckpoint =
+ alignWatermarkPerCheckpoint(checkpointId, subtaskCommittables, aligner);
+ if (checkpointId == END_INPUT_CHECKPOINT_ID && endInputWatermark != null) {
+ watermarkPerCheckpoint.put(END_INPUT_CHECKPOINT_ID, endInputWatermark);
+ }
+ return watermarkPerCheckpoint;
+ }
+
private void commitUpToCheckpoint(
long checkpointId,
Map toCommit,
@@ -620,18 +733,21 @@ public static class Provider extends RecreateOnResetOperatorCoordinator.Provider
private final boolean streamingCheckpointEnabled;
private final String initialCommitUser;
private final boolean failoverAfterRecovery;
+ @Nullable private final Long endInputWatermark;
public Provider(
OperatorID operatorId,
Committer.Factory committerFactory,
boolean streamingCheckpointEnabled,
String initialCommitUser,
- boolean failoverAfterRecovery) {
+ boolean failoverAfterRecovery,
+ @Nullable Long endInputWatermark) {
super(operatorId);
this.committerFactory = committerFactory;
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
this.initialCommitUser = initialCommitUser;
this.failoverAfterRecovery = failoverAfterRecovery;
+ this.endInputWatermark = endInputWatermark;
}
@Override
@@ -641,7 +757,8 @@ public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) {
committerFactory,
streamingCheckpointEnabled,
initialCommitUser,
- failoverAfterRecovery);
+ failoverAfterRecovery,
+ endInputWatermark);
}
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java
index 1562fbf4ee37..55dc78043cac 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java
@@ -37,16 +37,27 @@
public class WriterCommittables {
private static final Logger LOG = LoggerFactory.getLogger(WriterCommittables.class);
+ private static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
+ /** Maximum ordinary checkpoint reported by this subtask. End input is tracked separately. */
private long maxCheckpointId;
+
private final NavigableMap committablesPerCheckpoint;
+ private boolean endInput;
@VisibleForTesting
WriterCommittables(long maxCheckpointId, List entries) {
this.maxCheckpointId = maxCheckpointId;
this.committablesPerCheckpoint = new TreeMap<>();
+ this.endInput = false;
for (CheckpointCommittables entry : entries) {
- if (entry.checkpointId() > maxCheckpointId) {
+ if (entry.checkpointId() == END_INPUT_CHECKPOINT_ID) {
+ if (endInput) {
+ throw new IllegalStateException(
+ "Invalid input committables, duplicate end input entry");
+ }
+ endInput = true;
+ } else if (entry.checkpointId() > maxCheckpointId) {
throw new IllegalStateException(
"Invalid input committables, max checkpoint id should not be less than "
+ "checkpoint id of committables, max checkpoint is "
@@ -65,8 +76,12 @@ public class WriterCommittables {
@VisibleForTesting
WriterCommittables(CheckpointCommittables entry) {
- this.maxCheckpointId = entry.checkpointId();
+ // Use -1 for the end-input ID to avoid treating it as an ordinary checkpoint and allow
+ // subsequent ordinary checkpoints to be merged.
+ this.maxCheckpointId =
+ entry.checkpointId() == END_INPUT_CHECKPOINT_ID ? -1 : entry.checkpointId();
this.committablesPerCheckpoint = new TreeMap<>();
+ this.endInput = entry.checkpointId() == END_INPUT_CHECKPOINT_ID;
committablesPerCheckpoint.put(entry.checkpointId(), entry);
}
@@ -80,29 +95,44 @@ public NavigableMap getCommittablesBeforeCheckpoin
}
public void clearCommittablesBeforeCheckpoint(long checkpointId, boolean inclusive) {
- if (checkpointId > maxCheckpointId || (checkpointId == maxCheckpointId && inclusive)) {
+ if (checkpointId == END_INPUT_CHECKPOINT_ID && inclusive) {
reset();
- } else {
- committablesPerCheckpoint.headMap(checkpointId, inclusive).clear();
+ return;
+ }
+
+ committablesPerCheckpoint.headMap(checkpointId, inclusive).clear();
+ if (checkpointId > maxCheckpointId || (checkpointId == maxCheckpointId && inclusive)) {
+ maxCheckpointId = -1;
}
}
public void reset() {
maxCheckpointId = -1;
+ endInput = false;
committablesPerCheckpoint.clear();
}
public void mergeWith(WriterCommittables other) {
- if (other.maxCheckpointId <= maxCheckpointId) {
+ if (other.maxCheckpointId >= 0 && other.maxCheckpointId <= maxCheckpointId) {
throw new IllegalStateException(
"It must merge later checkpoint committables, however current checkpoint id is "
+ maxCheckpointId
+ ", to be merged checkpoint id is "
+ other.maxCheckpointId);
}
- maxCheckpointId = other.maxCheckpointId;
+ if (other.maxCheckpointId >= 0) {
+ maxCheckpointId = other.maxCheckpointId;
+ }
for (Map.Entry entry :
other.getCommittablesPerCheckpoint().entrySet()) {
+ if (entry.getKey() == END_INPUT_CHECKPOINT_ID) {
+ // End input may be replayed after failover. Each event is an authoritative
+ // snapshot for this subtask, so replace instead of treating it as a duplicate
+ // ordinary checkpoint or appending a second copy.
+ committablesPerCheckpoint.put(entry.getKey(), entry.getValue());
+ endInput = true;
+ continue;
+ }
if (committablesPerCheckpoint.containsKey(entry.getKey())) {
LOG.error(
"Subtask committables should not contain {}, the current committables are "
@@ -121,6 +151,18 @@ public long getMaxCheckpointId() {
return maxCheckpointId;
}
+ public boolean isEndInput() {
+ return endInput;
+ }
+
+ public boolean coversCheckpoint(long checkpointId) {
+ return endInput || maxCheckpointId >= checkpointId;
+ }
+
+ public CheckpointCommittables getEndInputCommittables() {
+ return committablesPerCheckpoint.get(END_INPUT_CHECKPOINT_ID);
+ }
+
/**
* Returns the watermark this subtask reported for {@code checkpointId}. Falls back to {@link
* Long#MIN_VALUE} when the subtask has no entry for that checkpoint — that "no observed
@@ -132,7 +174,14 @@ public long getMaxCheckpointId() {
*/
public long watermarkAt(long checkpointId) {
CheckpointCommittables entry = committablesPerCheckpoint.get(checkpointId);
- return entry == null ? Long.MIN_VALUE : entry.watermark();
+ if (entry != null) {
+ return entry.watermark();
+ }
+ // A finished subtask no longer constrains the watermark of later ordinary checkpoints.
+ // Returning MAX makes it neutral in the coordinator's min aggregation.
+ return endInput && checkpointId != END_INPUT_CHECKPOINT_ID
+ ? Long.MAX_VALUE
+ : Long.MIN_VALUE;
}
/**
@@ -148,8 +197,8 @@ public boolean isIdleAt(long checkpointId) {
@Override
public String toString() {
return String.format(
- "WriterCommittables{maxCheckpointId=%d, committables=%s}",
- maxCheckpointId, committablesPerCheckpoint);
+ "WriterCommittables{maxCheckpointId=%d, endInput=%s, committables=%s}",
+ maxCheckpointId, endInput, committablesPerCheckpoint);
}
public static WriterCommittables from(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorEndInputCommitITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorEndInputCommitITCase.java
new file mode 100644
index 000000000000..223a73e746f9
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorEndInputCommitITCase.java
@@ -0,0 +1,478 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+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.source.SplitListState;
+import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.CloseableIterator;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.Types;
+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.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.testutils.InMemoryReporter;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for end-input watermark on unaware-bucket append tables. */
+public class CoordinatorEndInputCommitITCase {
+
+ private static final int DEFAULT_PARALLELISM = 2;
+ private static final long WAIT_TIMEOUT_MILLIS = 60_000L;
+ private static final InMemoryReporter reporter = InMemoryReporter.create();
+
+ @RegisterExtension
+ protected static final org.apache.paimon.flink.util.MiniClusterWithClientExtension
+ MINI_CLUSTER_EXTENSION =
+ new org.apache.paimon.flink.util.MiniClusterWithClientExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+ .setConfiguration(
+ reporter.addToConfiguration(new Configuration()))
+ .build());
+
+ @TempDir Path tempPath;
+
+ @AfterEach
+ public final void cleanupRunningJobs() throws Exception {
+ ClusterClient clusterClient = MINI_CLUSTER_EXTENSION.createRestClusterClient();
+ for (JobStatusMessage job : clusterClient.listJobs().get()) {
+ if (!job.getJobState().isTerminalState()) {
+ try {
+ clusterClient.cancel(job.getJobId()).get(30, TimeUnit.SECONDS);
+ } catch (Exception ignored) {
+ // best-effort cleanup
+ }
+ }
+ }
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCoordinatorCommitEndInputInStreamingMode() throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ true,
+ true,
+ 12345L);
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testOperatorCommitEndInputInStreamingMode() throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ false,
+ true,
+ 12345L);
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testEndInputIsNotCommittedWithoutCheckpointsAfterTasksFinish() throws Exception {
+ assertEndInputNotCommittedWithoutCheckpointsAfterTasksFinish(true);
+ assertEndInputNotCommittedWithoutCheckpointsAfterTasksFinish(false);
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testEmptyFinalCommitDoesNotUpdateEndInputWatermarkInStreamingMode()
+ throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ true,
+ false,
+ 12345L);
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ false,
+ false,
+ 12345L);
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testForceCreateSnapshotUpdatesEndInputWatermarkInStreamingMode() throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ true,
+ false,
+ 12345L,
+ true);
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ false,
+ false,
+ 12345L,
+ true);
+ }
+
+ @Timeout(value = 120, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCoordinatorCommitEndInputInBatchMode() throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inBatchMode().build(),
+ false,
+ true,
+ false,
+ 12345L);
+ }
+
+ private void assertEndInputWatermark(
+ EnvironmentSettings settings,
+ boolean streaming,
+ boolean coordinatorCommit,
+ boolean emitAfterFirstCheckpoint,
+ long expectedWatermark)
+ throws Exception {
+ assertEndInputWatermark(
+ settings,
+ streaming,
+ coordinatorCommit,
+ emitAfterFirstCheckpoint,
+ expectedWatermark,
+ false);
+ }
+
+ private void assertEndInputWatermark(
+ EnvironmentSettings settings,
+ boolean streaming,
+ boolean coordinatorCommit,
+ boolean emitAfterFirstCheckpoint,
+ long expectedWatermark,
+ boolean forceCreateSnapshot)
+ throws Exception {
+ assertEndInputWatermark(
+ settings,
+ streaming,
+ coordinatorCommit,
+ emitAfterFirstCheckpoint,
+ expectedWatermark,
+ forceCreateSnapshot,
+ true);
+ }
+
+ private void assertEndInputWatermark(
+ EnvironmentSettings settings,
+ boolean streaming,
+ boolean coordinatorCommit,
+ boolean emitAfterFirstCheckpoint,
+ long expectedWatermark,
+ boolean forceCreateSnapshot,
+ boolean checkpointsAfterTasksFinish)
+ throws Exception {
+ final long endInputWatermark = 12345L;
+ String tableName = coordinatorCommit ? "T_END_INPUT_COORDINATOR" : "T_END_INPUT_OPERATOR";
+ StreamExecutionEnvironment streamEnv = null;
+ TableEnvironment tEnv;
+ if (streaming) {
+ streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ streamEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
+ streamEnv.setParallelism(DEFAULT_PARALLELISM);
+ streamEnv.enableCheckpointing(
+ checkpointsAfterTasksFinish ? 200L : TimeUnit.HOURS.toMillis(1));
+ Configuration configuration = new Configuration();
+ configuration.setString(
+ "execution.checkpointing.checkpoints-after-tasks-finish",
+ String.valueOf(checkpointsAfterTasksFinish));
+ if (!checkpointsAfterTasksFinish) {
+ configuration.setString("restart-strategy.type", "none");
+ }
+ streamEnv.configure(configuration);
+ tEnv = StreamTableEnvironment.create(streamEnv);
+ } else {
+ tEnv = TableEnvironment.create(settings);
+ }
+ tEnv.getConfig()
+ .getConfiguration()
+ .setString("table.exec.resource.default-parallelism", "2");
+
+ tEnv.executeSql(
+ "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' = '"
+ + tempPath
+ + "' )");
+ tEnv.executeSql("USE CATALOG mycat");
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tableName
+ + " (id INT, data STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'write-only' = 'true', "
+ + (coordinatorCommit ? "'sink.coordinator-commit.enabled' = 'true', " : "")
+ + (forceCreateSnapshot
+ ? "'"
+ + CoreOptions.COMMIT_FORCE_CREATE_SNAPSHOT.key()
+ + "' = 'true', "
+ : "")
+ + "'end-input.watermark' = '"
+ + endInputWatermark
+ + "')");
+ int expectedRowCount;
+ if (streaming) {
+ // Control the final-commit contents instead of relying on the timing between datagen
+ // completion and periodic checkpoints. Emitting after a completed checkpoint leaves a
+ // non-empty END_INPUT committable; emitting before it and ending only after it has
+ // completed leaves an empty one.
+ DataStream source =
+ checkpointsAfterTasksFinish
+ ? streamEnv
+ .fromSource(
+ emitAfterFirstCheckpoint
+ ? new EmitAfterFirstCheckpointSource()
+ : new EmitBeforeFirstCheckpointThenFinishSource(),
+ WatermarkStrategy.noWatermarks(),
+ "Controlled End Input Source")
+ .map(row -> row)
+ .returns(Types.ROW(Types.INT, Types.STRING))
+ .setParallelism(1)
+ : StreamExecutionEnvironmentUtils.fromData(
+ streamEnv,
+ Types.ROW(Types.INT, Types.STRING),
+ Row.of(1, "end-input"))
+ .setParallelism(1);
+ StreamTableEnvironment streamTableEnv = (StreamTableEnvironment) tEnv;
+ streamTableEnv.createTemporaryView("src", streamTableEnv.fromDataStream(source));
+ // The table planner restores the configured sink parallelism, so both source
+ // readers emit one final record after checkpoint 1.
+ expectedRowCount = DEFAULT_PARALLELISM;
+ } else {
+ tEnv.executeSql(
+ "CREATE TEMPORARY TABLE src (id INT, data STRING) WITH ("
+ + "'connector' = 'datagen', "
+ + "'number-of-rows' = '20', "
+ + "'rows-per-second' = '10', "
+ + "'fields.id.kind' = 'sequence', "
+ + "'fields.id.start' = '1', "
+ + "'fields.id.end' = '20', "
+ + "'fields.data.length' = '8')");
+ expectedRowCount = 20;
+ }
+
+ tEnv.executeSql("INSERT INTO " + tableName + " SELECT * FROM src").await();
+
+ FileStoreTable table =
+ (FileStoreTable)
+ ((FlinkCatalog) tEnv.getCatalog("mycat").get())
+ .catalog()
+ .getTable(Identifier.create("default", tableName));
+ if (!checkpointsAfterTasksFinish) {
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ return;
+ }
+ waitUntilRowCount(table, expectedRowCount);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(snapshot).isNotNull();
+ assertThat(snapshot.watermark()).isEqualTo(expectedWatermark);
+ }
+
+ private void assertEndInputNotCommittedWithoutCheckpointsAfterTasksFinish(
+ boolean coordinatorCommit) throws Exception {
+ assertEndInputWatermark(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ true,
+ coordinatorCommit,
+ true,
+ 12345L,
+ false,
+ false);
+ }
+
+ /** Emits the only record after checkpoint 1 has completed. */
+ private static class EmitAfterFirstCheckpointSource extends AbstractNonCoordinatedSource {
+
+ private static final long serialVersionUID = 1L;
+
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader createReader(SourceReaderContext context) {
+ return new Reader();
+ }
+
+ private static class Reader extends AbstractNonCoordinatedSourceReader {
+
+ private final SplitListState emittedState =
+ new SplitListState<>(
+ "emitted", value -> String.valueOf(value), Boolean::parseBoolean);
+ private boolean emitted;
+ private boolean firstCheckpointCompleted;
+
+ @Override
+ public InputStatus pollNext(ReaderOutput output) {
+ if (!firstCheckpointCompleted) {
+ return InputStatus.MORE_AVAILABLE;
+ }
+ if (!emitted) {
+ output.collect(Row.of(1, "end-input"));
+ emitted = true;
+ return InputStatus.MORE_AVAILABLE;
+ }
+ return InputStatus.END_OF_INPUT;
+ }
+
+ @Override
+ public void addSplits(List splits) {
+ emittedState.restoreState(splits);
+ for (Boolean state : emittedState.get()) {
+ emitted = state;
+ }
+ }
+
+ @Override
+ public List snapshotState(long checkpointId) {
+ emittedState.clear();
+ emittedState.add(emitted);
+ return emittedState.snapshotState();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ firstCheckpointCompleted = true;
+ }
+ }
+ }
+
+ /** Emits the only record before checkpoint 1, then ends only after it has completed. */
+ private static class EmitBeforeFirstCheckpointThenFinishSource
+ extends AbstractNonCoordinatedSource {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader createReader(SourceReaderContext context) {
+ return new Reader();
+ }
+
+ private static class Reader extends AbstractNonCoordinatedSourceReader {
+
+ private final SplitListState emittedState =
+ new SplitListState<>(
+ "emitted", value -> String.valueOf(value), Boolean::parseBoolean);
+ private boolean emitted;
+ private boolean firstCheckpointCompleted;
+
+ @Override
+ public InputStatus pollNext(ReaderOutput output) {
+ if (!emitted) {
+ output.collect(Row.of(1, "before-checkpoint"));
+ emitted = true;
+ return InputStatus.MORE_AVAILABLE;
+ }
+ if (!firstCheckpointCompleted) {
+ return InputStatus.MORE_AVAILABLE;
+ }
+ return InputStatus.END_OF_INPUT;
+ }
+
+ @Override
+ public void addSplits(List splits) {
+ emittedState.restoreState(splits);
+ for (Boolean state : emittedState.get()) {
+ emitted = state;
+ }
+ }
+
+ @Override
+ public List snapshotState(long checkpointId) {
+ emittedState.clear();
+ emittedState.add(emitted);
+ return emittedState.snapshotState();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ firstCheckpointCompleted = true;
+ }
+ }
+ }
+
+ private long readRowCount(FileStoreTable table) throws Exception {
+ RecordReader reader =
+ table.newRead().createReader(table.newSnapshotReader().read());
+ long rowCount = 0L;
+ try (CloseableIterator iterator = new RecordReaderIterator<>(reader)) {
+ while (iterator.hasNext()) {
+ iterator.next();
+ rowCount++;
+ }
+ }
+ return rowCount;
+ }
+
+ private void waitUntilRowCount(FileStoreTable table, long expectedRowCount) throws Exception {
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ long rowCount;
+ do {
+ rowCount = readRowCount(table);
+ if (rowCount == expectedRowCount) {
+ return;
+ }
+ Thread.sleep(100);
+ } while (System.currentTimeMillis() < deadline);
+
+ assertThat(rowCount).isEqualTo(expectedRowCount);
+ }
+}
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 bae18ea2c2bf..8a27a99fbb92 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
@@ -21,6 +21,7 @@
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.sink.coordinator.CheckpointCommittables;
import org.apache.paimon.flink.sink.coordinator.CheckpointCommittablesSerializer;
import org.apache.paimon.flink.sink.coordinator.CommittableEvent;
@@ -92,7 +93,8 @@ public void testWriterSendsCommittablesToCoordinatorAndStillEmitsDownstream() th
table, table.newCommit(context.commitUser()), context),
true,
commitUser,
- false);
+ false,
+ null);
coordinator.start();
coordinator.waitProcessAllActions();
@@ -443,6 +445,87 @@ public void testIdleFlagResetsToActiveOnRestore() throws Exception {
secondHarness.close();
}
+ @Test
+ public void testEndInputSendsMaxCheckpointAndForwardsCurrentWatermark() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(
+ options -> {
+ options.set(CoreOptions.BUCKET, -1);
+ options.remove("bucket-key");
+ options.set(FlinkConnectorOptions.END_INPUT_WATERMARK, 12345L);
+ });
+ String commitUser = UUID.randomUUID().toString();
+ List events = new ArrayList<>();
+ OneInputStreamOperatorTestHarness harness =
+ createHarness(table, commitUser, events::add);
+ harness.setup(new CommittableTypeInfo().createSerializer(new ExecutionConfig()));
+ harness.open();
+
+ harness.processElement(GenericRow.of(1, 10L), 1);
+ harness.endInput();
+
+ assertThat(events).hasSize(1);
+ CommittableEvent event = (CommittableEvent) events.get(0);
+ assertThat(event.getCheckpointId()).isEqualTo(Long.MAX_VALUE);
+ CheckpointCommittables entry = event.deserialize(COMMITTABLES_SERIALIZER);
+ assertThat(entry.checkpointId()).isEqualTo(Long.MAX_VALUE);
+ assertThat(entry.watermark()).isEqualTo(Long.MIN_VALUE);
+ assertThat(entry.committables()).hasSize(1);
+ CoordinatorCommittingRowDataStoreWriteOperator operator =
+ (CoordinatorCommittingRowDataStoreWriteOperator) harness.getOperator();
+ assertThat(operator.getPendingCommittables()).containsKey(Long.MAX_VALUE);
+
+ harness.close();
+ }
+
+ @Test
+ public void testRestoredEndInputSurvivesAnotherSnapshot() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ TypeSerializer serializer =
+ new CommittableTypeInfo().createSerializer(new ExecutionConfig());
+
+ List firstEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness first =
+ createHarness(table, commitUser, firstEvents::add);
+ first.setup(serializer);
+ first.open();
+ first.processElement(GenericRow.of(1, 10L), 1);
+ first.endInput();
+ OperatorSubtaskState firstSnapshot = first.snapshot(1L, 10L);
+ first.close();
+
+ List secondEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness second =
+ createHarness(table, commitUser, secondEvents::add);
+ second.setup(serializer);
+ restoreWithCheckpointId(second, firstSnapshot, 1L);
+ second.open();
+ CoordinatorCommittingRowDataStoreWriteOperator secondOperator =
+ (CoordinatorCommittingRowDataStoreWriteOperator) second.getOperator();
+ assertThat(secondOperator.getPendingCommittables()).containsKey(Long.MAX_VALUE);
+ RestoredCommittableEvent restored = (RestoredCommittableEvent) secondEvents.get(0);
+ assertThat(restored.deserialize(COMMITTABLES_SERIALIZER))
+ .extracting(CheckpointCommittables::checkpointId)
+ .contains(Long.MAX_VALUE);
+
+ OperatorSubtaskState secondSnapshot = second.snapshot(2L, 20L);
+ second.close();
+
+ List thirdEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness third =
+ createHarness(table, commitUser, thirdEvents::add);
+ third.setup(serializer);
+ restoreWithCheckpointId(third, secondSnapshot, 2L);
+ third.open();
+ RestoredCommittableEvent restoredAgain = (RestoredCommittableEvent) thirdEvents.get(0);
+ assertThat(restoredAgain.deserialize(COMMITTABLES_SERIALIZER))
+ .extracting(CheckpointCommittables::checkpointId)
+ .contains(Long.MAX_VALUE);
+
+ third.close();
+ }
+
private void assertCommittableEventCheckpoint(OperatorEvent event, long expectedCheckpointId) {
CommittableEvent committableEvent = (CommittableEvent) event;
assertThat(committableEvent.getCheckpointId()).isEqualTo(expectedCheckpointId);
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 91110d996218..00a26430f41c 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
@@ -48,16 +48,23 @@ public class FlinkSinkTest extends CommitterOperatorTestBase {
@Test
public void testCoordinatorCommitPreconditionsHappyPath() throws Exception {
FileStoreTable table = createUnawareBucketTable(options -> {});
- FlinkSink.checkCoordinatorCommitPreconditions(table, newCheckpointConfig(1), true);
+ FlinkSink.checkCoordinatorCommitPreconditions(table, newCheckpointConfig(1), true, true);
}
@Test
- public void testCoordinatorCommitPreconditionsRejectsBatchOrNoCheckpoint() throws Exception {
+ public void testCoordinatorCommitPreconditionsAllowsBatchWithoutCheckpoint() throws Exception {
+ FileStoreTable table = createUnawareBucketTable(options -> {});
+ FlinkSink.checkCoordinatorCommitPreconditions(table, newCheckpointConfig(1), false, false);
+ }
+
+ @Test
+ public void testCoordinatorCommitPreconditionsRejectsStreamingWithoutCheckpoint()
+ throws Exception {
FileStoreTable table = createUnawareBucketTable(options -> {});
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), false))
+ table, newCheckpointConfig(1), true, false))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -67,7 +74,7 @@ public void testCoordinatorCommitPreconditionsRejectsPrimaryKeyTable() throws Ex
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
+ table, newCheckpointConfig(1), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -77,7 +84,7 @@ public void testCoordinatorCommitPreconditionsRejectsFixedBucket() throws Except
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
+ table, newCheckpointConfig(1), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -88,7 +95,7 @@ public void testCoordinatorCommitPreconditionsRejectsNonWriteOnly() throws Excep
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
+ table, newCheckpointConfig(1), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -100,7 +107,7 @@ public void testCoordinatorCommitPreconditionsRejectsPrecommitCompact() throws E
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
+ table, newCheckpointConfig(1), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -114,7 +121,7 @@ public void testCoordinatorCommitPreconditionsRejectsAutoTagForSavepoint() throw
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(1), true))
+ table, newCheckpointConfig(1), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -124,7 +131,7 @@ public void testCoordinatorCommitPreconditionsRejectsConcurrentCheckpoints() thr
assertThatThrownBy(
() ->
FlinkSink.checkCoordinatorCommitPreconditions(
- table, newCheckpointConfig(2), true))
+ table, newCheckpointConfig(2), true, true))
.isInstanceOf(IllegalArgumentException.class);
}
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 a43bba18d3ad..0bf0e85b5837 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
@@ -60,6 +60,7 @@
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -98,6 +99,23 @@ public void checkNoFailure() {
assertThat(failureCause).isNull();
}
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testContextIsNotAccessedBeforeStart() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ DelayedInitializationContext context =
+ new DelayedInitializationContext(new OperatorID(), 2);
+
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ context.initialize();
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ assertThat(coordinator.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING);
+ coordinator.close();
+ }
+
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@Test
public void testCommitSingleSubtask() throws Exception {
@@ -207,6 +225,263 @@ public void testWatermarkAlignsMinAcrossSubtasks() throws Exception {
coordinator.close();
}
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testSubtasksEndInputAcrossDifferentCheckpoints() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, 1L, 1)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, 1L, 2)));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2");
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, Long.MAX_VALUE, 3)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, 2L, 4)));
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+ // The early end-input entry stays buffered while the other subtask is still running.
+ assertResults(table, "1, 1", "2, 2", "4, 4");
+
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 5)));
+ coordinator.waitProcessAllActions();
+ // Streaming mode still waits for a completed checkpoint before the final commit.
+ assertResults(table, "1, 1", "2, 2", "4, 4");
+
+ coordinator.notifyCheckpointComplete(3L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2", "3, 3", "4, 4", "5, 5");
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testRepeatedEndInputEventIsIdempotent() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ CommittableEvent repeated = event(committable(table, Long.MAX_VALUE, 1));
+ coordinator.handleEventFromOperator(0, 0, repeated);
+ coordinator.handleEventFromOperator(0, 0, repeated);
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 2)));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+
+ assertThat(failureCause).isNull();
+ assertResults(table, "1, 1", "2, 2");
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCheckpointDisabledCommitsWhenAllSubtasksEndInput() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator =
+ new CommittingWriteOperatorCoordinator(
+ context,
+ commitContext ->
+ new StoreCommitter(
+ table,
+ table.newStreamWriteBuilder()
+ .withCommitUser(commitContext.commitUser())
+ .newCommit(),
+ commitContext),
+ false,
+ commitUser,
+ false,
+ null);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, Long.MAX_VALUE, 1)));
+ coordinator.waitProcessAllActions();
+ assertThat(table.latestSnapshot()).isNotPresent();
+
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 2)));
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2");
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCloseDrainsCheckpointDisabledEndInputCommit() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator =
+ new CommittingWriteOperatorCoordinator(
+ context,
+ commitContext ->
+ new StoreCommitter(
+ table,
+ table.newStreamWriteBuilder()
+ .withCommitUser(commitContext.commitUser())
+ .newCommit(),
+ commitContext),
+ false,
+ commitUser,
+ false,
+ null);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, Long.MAX_VALUE, 1)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 2)));
+ // Do not add a test fence here. close() itself must wait for both queued events and the
+ // final commit before closing the committer.
+ coordinator.close();
+
+ assertResults(table, "1, 1", "2, 2");
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testRepeatedEndInputAfterFinalCommitIsIgnored() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, Long.MAX_VALUE, 1)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 2)));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2");
+
+ coordinator.handleEventFromOperator(0, 1, event(committable(table, Long.MAX_VALUE, 3)));
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+
+ assertThat(failureCause).isNull();
+ assertResults(table, "1, 1", "2, 2");
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testPartialFailoverReplaysEndInputWhileCoordinatorIsRunning() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ Committable earlyEndInput = committable(table, Long.MAX_VALUE, 1);
+ coordinator.handleEventFromOperator(0, 0, event(earlyEndInput));
+ coordinator.waitProcessAllActions();
+
+ coordinator.subtaskReset(0, 1L);
+ coordinator.handleEventFromOperator(
+ 0,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(earlyEndInput),
+ Long.MIN_VALUE)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 2)));
+ coordinator.waitProcessAllActions();
+ // Streaming mode must still wait for a completed checkpoint after replay reaches global
+ // EndInput.
+ assertThat(table.latestSnapshot()).isNotPresent();
+
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2");
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testPartialFailoverWhileAllSubtasksEndInputWaitForCheckpointComplete()
+ throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ // Subtask 0 reaches EndInput before checkpoint 1. Subtask 1 is still running, so completing
+ // checkpoint 1 commits only its ordinary entry and keeps subtask 0's MAX entry pending.
+ Committable earlyEndInput = committable(table, Long.MAX_VALUE, 1);
+ coordinator.handleEventFromOperator(0, 0, event(earlyEndInput));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, 1L, 2)));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "2, 2");
+ assertThat(table.snapshotManager().snapshotCount()).isEqualTo(1);
+
+ // The last running subtask now reaches EndInput. All subtasks are at EndInput, but
+ // streaming mode must wait for the next completed checkpoint before the final commit.
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 3)));
+ coordinator.waitProcessAllActions();
+ assertResults(table, "2, 2");
+ assertThat(table.snapshotManager().snapshotCount()).isEqualTo(1);
+
+ // While waiting for checkpoint completion, subtask 0 fails and restores its MAX entry from
+ // checkpoint 1. Rebuilding allSubtasksEndInput must not commit eagerly in streaming mode.
+ coordinator.executionAttemptFailed(0, 0, new Exception("Fail subtask 0 as expected"));
+ coordinator.executionAttemptReady(0, 1, new MockSubtaskGateway());
+ coordinator.subtaskReset(0, 1L);
+ coordinator.waitProcessAllActions();
+ coordinator.handleEventFromOperator(
+ 0,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(earlyEndInput),
+ Long.MIN_VALUE)));
+ coordinator.waitProcessAllActions();
+
+ assertThat(failureCause).isNull();
+ assertResults(table, "2, 2");
+ assertThat(table.snapshotManager().snapshotCount()).isEqualTo(1);
+
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+ assertThat(failureCause).isNull();
+ assertResults(table, "1, 1", "2, 2", "3, 3");
+ assertThat(table.snapshotManager().snapshotCount()).isEqualTo(2);
+ coordinator.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCheckpointAbortPreservesEarlyEndInput() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(0, 0, event(committable(table, Long.MAX_VALUE, 1)));
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, 1L, 2)));
+ coordinator.notifyCheckpointAborted(1L);
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, 2L, 3)));
+ coordinator.notifyCheckpointComplete(2L);
+ coordinator.waitProcessAllActions();
+ // The later completed checkpoint commits ordinary entries, but not the early MAX entry.
+ assertResults(table, "2, 2", "3, 3");
+
+ coordinator.handleEventFromOperator(1, 0, event(committable(table, Long.MAX_VALUE, 4)));
+ coordinator.notifyCheckpointComplete(3L);
+ coordinator.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2", "3, 3", "4, 4");
+ coordinator.close();
+ }
+
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@Test
public void testRestoringAlignsBeforeRunning() throws Exception {
@@ -253,6 +528,265 @@ public void testRestoringAlignsBeforeRunning() throws Exception {
second.close();
}
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testWriterFailoverThenGlobalRestoreUsesCheckpointedEndInput() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+
+ Committable checkpointedSubtaskZero = committable(table, Long.MAX_VALUE, 1);
+ Committable checkpointedSubtaskOne = committable(table, Long.MAX_VALUE, 2);
+
+ // The coordinator checkpoint is persisted, but its completion callback is lost when the
+ // JM fails. Both writers have already snapshotted EndInput into this checkpoint.
+ CommittingWriteOperatorCoordinator first = createCoordinator(table, context, true);
+ first.start();
+ first.waitProcessAllActions();
+ CompletableFuture checkpoint = new CompletableFuture<>();
+ first.checkpointCoordinator(1L, checkpoint);
+ first.handleEventFromOperator(0, 0, event(checkpointedSubtaskZero));
+ first.handleEventFromOperator(1, 0, event(checkpointedSubtaskOne));
+ first.waitProcessAllActions();
+ byte[] state = checkpoint.get();
+ assertResults(table);
+
+ // Before the JM fails, writer-0 fails and restores locally. This replay only mutates the
+ // old coordinator instance and must not become recovery authority for the new instance.
+ first.executionAttemptFailed(0, 0, new Exception("Fail subtask 0 as expected"));
+ first.executionAttemptReady(0, 1, new MockSubtaskGateway());
+ first.subtaskReset(0, 1L);
+ first.waitProcessAllActions();
+ first.handleEventFromOperator(0, 1, restoreEvent(1L, checkpointedSubtaskZero));
+ first.waitProcessAllActions();
+ assertResults(table);
+ first.close();
+
+ // JM failover creates a fresh coordinator. All writers recover from the same completed
+ // checkpoint, so only the checkpointed EndInput committables are eligible for final commit.
+ CommittingWriteOperatorCoordinator second = createCoordinator(table, context, true);
+ second.resetToCheckpoint(1L, state);
+ second.start();
+ second.waitProcessAllActions();
+ second.handleEventFromOperator(
+ 0,
+ 2,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(checkpointedSubtaskZero),
+ Long.MIN_VALUE)));
+ second.waitProcessAllActions();
+ assertThat(second.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING);
+ assertResults(table);
+
+ second.handleEventFromOperator(
+ 1,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(checkpointedSubtaskOne),
+ Long.MIN_VALUE)));
+ second.waitProcessAllActions();
+
+ assertResults(table, "1, 1", "2, 2");
+ assertThat(failureCause).isInstanceOf(RuntimeException.class);
+ assertThat(failureCause).hasMessageContaining("intentionally thrown");
+ failureCause = null;
+ second.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testWriterFailoverThenGlobalRestoreRetainsPartialCheckpointedEndInput()
+ throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+
+ Committable earlyEndInput = committable(table, Long.MAX_VALUE, 1);
+ Committable checkpointedRunningWriter = committable(table, 1L, 2);
+
+ // Checkpoint 1 contains EndInput for writer-0 while writer-1 is still running. Its
+ // completion commits only writer-1's ordinary entry and leaves writer-0's MAX pending.
+ CommittingWriteOperatorCoordinator first = createCoordinator(table, context, true);
+ first.start();
+ first.waitProcessAllActions();
+ CompletableFuture checkpoint = new CompletableFuture<>();
+ first.checkpointCoordinator(1L, checkpoint);
+ first.handleEventFromOperator(0, 0, event(earlyEndInput));
+ first.handleEventFromOperator(1, 0, event(checkpointedRunningWriter));
+ first.notifyCheckpointComplete(1L);
+ first.waitProcessAllActions();
+ byte[] state = checkpoint.get();
+ assertResults(table, "2, 2");
+
+ // Writer-0 then fails and replays its checkpointed MAX to the old running coordinator.
+ // A following JM failover must discard this local recovery progress.
+ first.executionAttemptFailed(0, 0, new Exception("Fail subtask 0 as expected"));
+ first.executionAttemptReady(0, 1, new MockSubtaskGateway());
+ first.subtaskReset(0, 1L);
+ first.waitProcessAllActions();
+ first.handleEventFromOperator(
+ 0,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(earlyEndInput),
+ Long.MIN_VALUE)));
+ first.waitProcessAllActions();
+ assertResults(table, "2, 2");
+ first.close();
+
+ // Global recovery restarts every writer from checkpoint 1. Only writer-0 restores
+ // EndInput, so the new coordinator recovers the ordinary target and keeps MAX pending.
+ CommittingWriteOperatorCoordinator second = createCoordinator(table, context, true);
+ second.resetToCheckpoint(1L, state);
+ second.start();
+ second.waitProcessAllActions();
+ second.handleEventFromOperator(
+ 0,
+ 2,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(earlyEndInput),
+ Long.MIN_VALUE)));
+ second.waitProcessAllActions();
+ assertThat(second.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING);
+
+ second.handleEventFromOperator(
+ 1,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ 1L,
+ Collections.singletonList(checkpointedRunningWriter),
+ Long.MIN_VALUE)));
+ second.waitProcessAllActions();
+ assertThat(second.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING);
+ assertThat(failureCause).isNull();
+ assertResults(table, "2, 2");
+
+ // The retained MAX is committed only after writer-1 reaches EndInput and a later streaming
+ // checkpoint completes.
+ second.handleEventFromOperator(1, 1, event(committable(table, Long.MAX_VALUE, 3)));
+ second.waitProcessAllActions();
+ assertResults(table, "2, 2");
+ second.notifyCheckpointComplete(2L);
+ second.waitProcessAllActions();
+ assertThat(failureCause).isNull();
+ assertResults(table, "1, 1", "2, 2", "3, 3");
+ second.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testGlobalRestoreRetainsPartialEndInput() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+
+ Committable checkpointOneSubtaskZero = committable(table, 1L, 1);
+ Committable checkpointOneSubtaskOne = committable(table, 1L, 2);
+ CommittingWriteOperatorCoordinator first = createCoordinator(table, context, false);
+ first.start();
+ first.waitProcessAllActions();
+ first.handleEventFromOperator(0, 0, event(checkpointOneSubtaskZero));
+ first.handleEventFromOperator(1, 0, event(checkpointOneSubtaskOne));
+ CompletableFuture checkpoint = new CompletableFuture<>();
+ first.checkpointCoordinator(1L, checkpoint);
+ first.notifyCheckpointComplete(1L);
+ first.waitProcessAllActions();
+ byte[] state = checkpoint.get();
+ first.close();
+ assertResults(table, "1, 1", "2, 2");
+
+ Committable earlyEndInput = committable(table, Long.MAX_VALUE, 3);
+ CommittingWriteOperatorCoordinator second = createCoordinator(table, context, true);
+ second.resetToCheckpoint(1L, state);
+ second.start();
+ second.waitProcessAllActions();
+ second.handleEventFromOperator(
+ 0,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ 1L,
+ Collections.singletonList(checkpointOneSubtaskZero),
+ Long.MIN_VALUE),
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(earlyEndInput),
+ Long.MIN_VALUE)));
+ second.handleEventFromOperator(
+ 1,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ 1L,
+ Collections.singletonList(checkpointOneSubtaskOne),
+ Long.MIN_VALUE)));
+ second.waitProcessAllActions();
+
+ assertThat(second.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING);
+ assertThat(failureCause).isNull();
+ assertResults(table, "1, 1", "2, 2");
+
+ second.handleEventFromOperator(1, 1, event(committable(table, Long.MAX_VALUE, 4)));
+ second.notifyCheckpointComplete(2L);
+ second.waitProcessAllActions();
+ assertResults(table, "1, 1", "2, 2", "3, 3", "4, 4");
+ second.close();
+ }
+
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testGlobalRestoreCommitsWhenAllSubtasksReachedEndInput() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, true);
+ coordinator.resetToCheckpoint(1L, emptyState());
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ coordinator.handleEventFromOperator(
+ 0,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(committable(table, Long.MAX_VALUE, 1)),
+ Long.MIN_VALUE)));
+ coordinator.handleEventFromOperator(
+ 1,
+ 1,
+ restoreEventOfEntries(
+ 1L,
+ new CheckpointCommittables(
+ Long.MAX_VALUE,
+ Collections.singletonList(committable(table, Long.MAX_VALUE, 2)),
+ Long.MIN_VALUE)));
+ coordinator.waitProcessAllActions();
+
+ assertResults(table, "1, 1", "2, 2");
+ assertThat(failureCause).isInstanceOf(RuntimeException.class);
+ assertThat(failureCause).hasMessageContaining("intentionally thrown");
+ failureCause = null;
+ coordinator.close();
+ }
+
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@Test
public void testSnapshotLostWhenFailed() throws Exception {
@@ -431,7 +965,8 @@ public void testCheckpointFutureCompletedExceptionallyOnSnapshotFailure() throws
expected),
true,
commitUser,
- false);
+ false,
+ null);
coordinator.start();
coordinator.waitProcessAllActions();
@@ -1125,7 +1660,8 @@ private CommittingWriteOperatorCoordinator createCoordinator(
commitContext),
true,
commitUser,
- failoverAfterRecovery);
+ failoverAfterRecovery,
+ null);
}
private CommittingWriteOperatorCoordinator createCoordinatorCapturingContext(
@@ -1145,7 +1681,8 @@ private CommittingWriteOperatorCoordinator createCoordinatorCapturingContext(
},
true,
commitUser,
- false);
+ false,
+ null);
}
private Committable committable(FileStoreTable table, long checkpointId, int value)
@@ -1234,6 +1771,12 @@ private RestoredCommittableEvent restoreEventOf(
restoredCheckpointId, Collections.singletonList(entry), SERIALIZER);
}
+ private RestoredCommittableEvent restoreEventOfEntries(
+ long restoredCheckpointId, CheckpointCommittables... entries) throws Exception {
+ return RestoredCommittableEvent.create(
+ restoredCheckpointId, Arrays.asList(entries), SERIALIZER);
+ }
+
private byte[] emptyState() throws Exception {
return SimpleVersionedSerialization.writeVersionAndSerialize(
new CoordinatorStateSerializer(),
@@ -1297,6 +1840,35 @@ public CheckpointCoordinator getCheckpointCoordinator() {
}
}
+ private class DelayedInitializationContext extends TestingContext {
+
+ private boolean initialized;
+
+ private DelayedInitializationContext(OperatorID operatorID, int parallelism) {
+ super(operatorID, parallelism);
+ }
+
+ private void initialize() {
+ initialized = true;
+ }
+
+ @Override
+ public int currentParallelism() {
+ if (!initialized) {
+ throw new IllegalStateException("Context was not yet initialized");
+ }
+ return super.currentParallelism();
+ }
+
+ @Override
+ public ClassLoader getUserCodeClassloader() {
+ if (!initialized) {
+ throw new IllegalStateException("Context was not yet initialized");
+ }
+ return super.getUserCodeClassloader();
+ }
+ }
+
/** {@link Committer} decorator whose {@link #snapshotState()} always throws. */
private static class FailingSnapshotCommitter
implements Committer {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java
index aa2b20c83247..b59270148943 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java
@@ -254,6 +254,68 @@ public void testInvalidMerge() {
assertThrows(IllegalStateException.class, () -> committables.mergeWith(oldCommittables));
}
+ @Test
+ public void testEndInputCoversLaterCheckpointAndDoesNotConstrainWatermark() {
+ CheckpointCommittables endInput =
+ new CheckpointCommittables(Long.MAX_VALUE, Collections.emptyList(), 1234L);
+ WriterCommittables committables = new WriterCommittables(endInput);
+
+ assertThat(committables.isEndInput()).isTrue();
+ assertThat(committables.getMaxCheckpointId()).isEqualTo(-1L);
+ assertThat(committables.coversCheckpoint(100L)).isTrue();
+ assertThat(committables.watermarkAt(100L)).isEqualTo(Long.MAX_VALUE);
+ assertThat(committables.watermarkAt(Long.MAX_VALUE)).isEqualTo(1234L);
+ }
+
+ @Test
+ public void testRestoreMayContainEndInputNewerThanRestoredCheckpoint() {
+ CheckpointCommittables ordinary =
+ new CheckpointCommittables(2L, Collections.emptyList(), 100L);
+ CheckpointCommittables endInput =
+ new CheckpointCommittables(Long.MAX_VALUE, Collections.emptyList(), 200L);
+
+ WriterCommittables committables =
+ new WriterCommittables(2L, Arrays.asList(ordinary, endInput));
+
+ assertThat(committables.getMaxCheckpointId()).isEqualTo(2L);
+ assertThat(committables.isEndInput()).isTrue();
+ assertThat(committables.getEndInputCommittables()).isSameAs(endInput);
+ }
+
+ @Test
+ public void testRepeatedEndInputReplacesAuthoritativeEntry() {
+ CheckpointCommittables first =
+ new CheckpointCommittables(Long.MAX_VALUE, Collections.emptyList(), 100L);
+ CheckpointCommittables second =
+ new CheckpointCommittables(Long.MAX_VALUE, Collections.emptyList(), 200L);
+ WriterCommittables committables = new WriterCommittables(first);
+
+ committables.mergeWith(new WriterCommittables(second));
+
+ assertThat(committables.isEndInput()).isTrue();
+ assertThat(committables.getEndInputCommittables()).isSameAs(second);
+ assertThat(committables.getCommittablesPerCheckpoint()).hasSize(1);
+ }
+
+ @Test
+ public void testFiniteClearKeepsEndInputUntilFinalCommit() {
+ WriterCommittables committables =
+ new WriterCommittables(
+ 2L,
+ Arrays.asList(
+ new CheckpointCommittables(2L, Collections.emptyList(), 100L),
+ new CheckpointCommittables(
+ Long.MAX_VALUE, Collections.emptyList(), 200L)));
+
+ committables.clearCommittablesBeforeCheckpoint(2L, true);
+ assertThat(committables.isEndInput()).isTrue();
+ assertThat(committables.getCommittablesPerCheckpoint()).containsOnlyKeys(Long.MAX_VALUE);
+
+ committables.clearCommittablesBeforeCheckpoint(Long.MAX_VALUE, true);
+ assertThat(committables.isEndInput()).isFalse();
+ assertThat(committables.getCommittablesPerCheckpoint()).isEmpty();
+ }
+
@Test
public void testClearCommittablesBeforeCheckpoint() {
CommitMessage commitMessage = createEmptyCommitMessage();