From 8a88dde6b32428b10ade5a14d6a542722886f640 Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Thu, 16 Jul 2026 00:18:39 +0800 Subject: [PATCH 1/8] [flink] Support end input in coordinator commit --- .../flink_connector_configuration.html | 2 +- .../paimon/flink/FlinkConnectorOptions.java | 5 +- ...orCommittingRowDataStoreWriteOperator.java | 29 +++++- .../apache/paimon/flink/sink/FlinkSink.java | 20 ++-- .../CommittingWriteOperatorCoordinator.java | 95 ++++++++++++++++--- .../sink/coordinator/WriterCommittables.java | 68 +++++++++++-- .../paimon/flink/CoordinatorCommitITCase.java | 63 ++++++++++++ ...mmittingRowDataStoreWriteOperatorTest.java | 82 ++++++++++++++++ .../paimon/flink/sink/FlinkSinkTest.java | 25 +++-- ...ommittingWriteOperatorCoordinatorTest.java | 85 +++++++++++++++++ .../coordinator/WriterCommittablesTest.java | 62 ++++++++++++ 11 files changed, 491 insertions(+), 45 deletions(-) diff --git a/docs/generated/flink_connector_configuration.html b/docs/generated/flink_connector_configuration.html index 83e9c0bb25c1..d09bca0d3f5c 100644 --- a/docs/generated/flink_connector_configuration.html +++ b/docs/generated/flink_connector_configuration.html @@ -270,7 +270,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 bc61b8bbab7a..2a555b9235e1 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 @@ -494,8 +494,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 1ff8ab06dda4..2b70c494fb8b 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 @@ -24,6 +24,7 @@ import org.apache.paimon.flink.sink.coordinator.CommittableEvent; import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator; import org.apache.paimon.flink.sink.coordinator.RestoredCommittableEvent; +import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.CommitMessageSerializer; import org.apache.paimon.utils.Preconditions; @@ -49,6 +50,8 @@ import java.util.NavigableMap; import java.util.TreeMap; +import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK; + /** * Write operator that hands committables to a JM-side {@link CommittingWriteOperatorCoordinator} * which performs the commit. @@ -64,6 +67,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"; @@ -79,6 +83,8 @@ public class CoordinatorCommittingRowDataStoreWriteOperator /** Latest watermark observed on the input; forwarded on subsequent events. */ private transient long currentWatermark; + private final Long endInputWatermark; + private transient CheckpointCommittablesSerializer stateSerializer; private transient TypeSerializer eventSerializer; @@ -90,6 +96,7 @@ public CoordinatorCommittingRowDataStoreWriteOperator( OperatorEventGateway operatorEventGateway) { super(parameters, table, storeSinkWriteProvider, initialCommitUser); this.operatorEventGateway = Preconditions.checkNotNull(operatorEventGateway); + this.endInputWatermark = new Options(table.options()).get(END_INPUT_WATERMARK); } @Override @@ -123,6 +130,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(); @@ -158,8 +170,23 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception { @Override protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException { List committables = prepareCommit(waitCompaction, checkpointId); + long watermark = + checkpointId == END_INPUT_CHECKPOINT_ID && endInputWatermark != null + ? endInputWatermark + : currentWatermark; CheckpointCommittables entry = - new CheckpointCommittables(checkpointId, committables, currentWatermark); + new CheckpointCommittables(checkpointId, committables, watermark); + + 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, watermark); + } + } // 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 7405ae4894e3..3872637aab3b 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 @@ -191,14 +191,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); } @@ -206,8 +208,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. @@ -357,15 +361,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 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 f57e4ec149fd..fc78720dcf2f 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 @@ -72,6 +72,7 @@ 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; @@ -93,6 +94,7 @@ public class CommittingWriteOperatorCoordinator implements OperatorCoordinator { private Committer committer; private String commitUser; private MemoryBackendStateStore stateStore; + private boolean endInputCommitted; public CommittingWriteOperatorCoordinator( OperatorCoordinator.Context context, @@ -117,6 +119,7 @@ public CommittingWriteOperatorCoordinator( Executors.newSingleThreadExecutor( new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context)); this.state = State.CREATED; + this.endInputCommitted = false; } @Override @@ -190,7 +193,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); } }, @@ -209,28 +211,44 @@ 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); + alignWatermarkPerCheckpoint(targetCheckpointId, subtaskCommittables); 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); @@ -297,8 +315,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); @@ -311,16 +336,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 " @@ -340,13 +379,41 @@ 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 = + alignWatermarkPerCheckpoint(END_INPUT_CHECKPOINT_ID, subtaskCommittables); + 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) { 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 243969069bcb..293b5e6b1c0b 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,11 @@ public class WriterCommittables { @VisibleForTesting WriterCommittables(CheckpointCommittables entry) { - this.maxCheckpointId = entry.checkpointId(); + // ? 为什么要 -1 + 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 +94,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 +150,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,14 +173,21 @@ 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; } @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/CoordinatorCommitITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java index 0ddbfbd18c48..1908ae96c26c 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java @@ -18,6 +18,7 @@ package org.apache.paimon.flink; +import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.InternalRow; import org.apache.paimon.reader.RecordReader; @@ -110,6 +111,68 @@ public void testCoordinatorCommitMetricsAndCommittedRows() throws Exception { assertThat(readRowCount(runningJob.table)).isGreaterThan(0L); } + @Timeout(value = 120, unit = TimeUnit.SECONDS) + @Test + public void testCoordinatorCommitEndInputInStreamingMode() throws Exception { + assertCoordinatorCommitEndInput( + EnvironmentSettings.newInstance().inStreamingMode().build(), true); + } + + @Timeout(value = 120, unit = TimeUnit.SECONDS) + @Test + public void testCoordinatorCommitEndInputInBatchMode() throws Exception { + assertCoordinatorCommitEndInput( + EnvironmentSettings.newInstance().inBatchMode().build(), false); + } + + private void assertCoordinatorCommitEndInput(EnvironmentSettings settings, boolean streaming) + throws Exception { + final long endInputWatermark = 12345L; + TableEnvironment tEnv = TableEnvironment.create(settings); + tEnv.getConfig() + .getConfiguration() + .setString("table.exec.resource.default-parallelism", "2"); + if (streaming) { + tEnv.getConfig() + .getConfiguration() + .setString("execution.checkpointing.interval", "200 ms"); + } + + tEnv.executeSql( + "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' = '" + + tempPath + + "' )"); + tEnv.executeSql("USE CATALOG mycat"); + tEnv.executeSql( + "CREATE TABLE T_END_INPUT (id INT, data STRING) WITH (" + + "'bucket' = '-1', " + + "'write-only' = 'true', " + + "'sink.coordinator-commit.enabled' = 'true', " + + "'end-input.watermark' = '" + + endInputWatermark + + "')"); + 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')"); + + tEnv.executeSql("INSERT INTO T_END_INPUT SELECT * FROM src").await(); + + FileStoreTable table = + (FileStoreTable) + ((FlinkCatalog) tEnv.getCatalog("mycat").get()) + .catalog() + .getTable(Identifier.create("default", "T_END_INPUT")); + assertThat(readRowCount(table)).isEqualTo(20L); + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot.watermark()).isEqualTo(endInputWatermark); + } + private RunningJob startStreamingInsert(boolean coordinatorCommitEnabled) throws Exception { String tableName = coordinatorCommitEnabled ? "T_COORDINATOR_COMMIT" : "T_DEFAULT_COMMIT"; TableEnvironment tEnv = 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 296eaa382344..f4fceadee12b 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; @@ -353,6 +354,87 @@ public void testEmptyRestoreStillSendsRestoreEvent() throws Exception { secondHarness.close(); } + @Test + public void testEndInputSendsMaxCheckpointAndUsesConfiguredWatermark() 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(12345L); + 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 5889f41cf52a..6e62e6847dc9 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 @@ -207,6 +207,91 @@ 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); + 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 testRestoringAlignsBeforeRunning() throws Exception { 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 74352dbed41a..08995d0e3c87 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(); From 5ab8315fd9a0d415fa217867f3124f734dc6178d Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Thu, 16 Jul 2026 17:28:04 +0800 Subject: [PATCH 2/8] Fix for batch end-input commit and coordinator initialization - Defer coordinator Context initialization - Pass actual streamingCheckpointEnabled from FlinkSink - Sync for END_INPUT event completion in batch mode - Remove unnecessary empty snapshot check in streaming IT --- .../apache/paimon/flink/sink/FlinkSink.java | 16 +++- .../paimon/flink/sink/RowAppendTableSink.java | 12 ++- .../CommittingWriteOperatorCoordinator.java | 84 ++++++++++++++++--- .../sink/coordinator/WriterCommittables.java | 3 +- .../paimon/flink/CoordinatorCommitITCase.java | 5 +- ...ommittingWriteOperatorCoordinatorTest.java | 46 ++++++++++ 6 files changed, 147 insertions(+), 19 deletions(-) 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 3872637aab3b..cfe20fd3d7bb 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 @@ -128,6 +128,8 @@ public DataStream doWrite( DataStream input, String commitUser, @Nullable Integer parallelism) { StreamExecutionEnvironment env = input.getExecutionEnvironment(); boolean isStreaming = isStreaming(input); + boolean streamingCheckpointEnabled = + isStreaming && env.getCheckpointConfig().isCheckpointingEnabled(); boolean writeOnly = table.coreOptions().writeOnly(); SingleOutputStreamOperator written = @@ -141,7 +143,8 @@ public DataStream doWrite( isStreaming, ignorePreviousFiles, hasSinkMaterializer(input)), - commitUser)); + commitUser, + streamingCheckpointEnabled)); if (parallelism == null) { forwardParallelism(written, input); } else { @@ -334,6 +337,17 @@ public static void assertBatchAdaptiveParallelism( endInputWatermark); } + /** + * Creates the writer operator factory with the runtime checkpoint mode when the writer needs + * it. Existing sinks do not use this information and keep the two-argument implementation. + */ + protected OneInputStreamOperatorFactory createWriteOperatorFactory( + StoreSinkWrite.Provider writeProvider, + String commitUser, + boolean streamingCheckpointEnabled) { + return createWriteOperatorFactory(writeProvider, commitUser); + } + protected abstract OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser); 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..9f37eec6cdfa 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 @@ -48,14 +48,20 @@ public RowAppendTableSink( @Override protected OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser) { + return createWriteOperatorFactory(writeProvider, commitUser, true); + } + + @Override + protected OneInputStreamOperatorFactory createWriteOperatorFactory( + StoreSinkWrite.Provider writeProvider, + String commitUser, + boolean streamingCheckpointEnabled) { 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()); } 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 fc78720dcf2f..6a819c77d7b0 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 @@ -50,6 +50,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; @@ -78,12 +79,12 @@ public class CommittingWriteOperatorCoordinator implements OperatorCoordinator { 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; // Populated by resetToCheckpoint and consumed by start. Plain fields are sufficient: both // callbacks run on the same scheduler thread in order. @@ -107,17 +108,12 @@ public CommittingWriteOperatorCoordinator( this.streamingCheckpointEnabled = streamingCheckpointEnabled; this.commitUser = initialCommitUser; this.failoverAfterRecovery = failoverAfterRecovery; - this.parallelism = context.currentParallelism(); - this.subtaskCommittables = new WriterCommittables[parallelism]; this.committablesSerializer = new SimpleVersionedSerializerTypeSerializerProxy<>( () -> new CheckpointCommittablesSerializer( new CommittableSerializer(new CommitMessageSerializer()))); this.stateSerializer = new CoordinatorStateSerializer(); - this.commitExecutor = - Executors.newSingleThreadExecutor( - new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context)); this.state = State.CREATED; this.endInputCommitted = false; } @@ -130,6 +126,14 @@ 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()]; + coordinatorThreadFactory = + new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context); + commitExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory); runInEventLoop( () -> { if (state == State.RESTORING) { @@ -186,7 +190,11 @@ public void checkpointCoordinator(long checkpointId, CompletableFuture r @Override public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) { - runInEventLoop( + boolean waitForBatchEndInputCommit = + !streamingCheckpointEnabled + && event instanceof CommittableEvent + && ((CommittableEvent) event).getCheckpointId() == END_INPUT_CHECKPOINT_ID; + ThrowingRunnable eventHandler = () -> { if (event instanceof CommittableEvent) { handleCommittableEvent(subtask, (CommittableEvent) event); @@ -195,11 +203,61 @@ public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEven } else { throw new UnsupportedOperationException("Unsupported event type: " + event); } + }; + + if (waitForBatchEndInputCommit) { + runInEventLoopSync( + eventHandler, + "handling operator event %s from subtask %d (#%d)", + event, + subtask, + attemptNumber); + } else { + runInEventLoop( + eventHandler, + "handling operator event %s from subtask %d (#%d)", + event, + subtask, + attemptNumber); + } + } + + /** + * Runs an action in the commit event loop and waits for completion. This is only used for the + * final end-input event in batch mode without checkpointing. It prevents the asynchronous + * coordinator event handling from returning before the final commit has completed, which could + * otherwise let the batch job finish prematurely. + */ + private void runInEventLoopSync( + ThrowingRunnable action, + String actionName, + Object... actionNameFormatParameters) { + checkState( + coordinatorThreadFactory == null + || !coordinatorThreadFactory.isCurrentThreadCoordinatorThread(), + "Cannot synchronously wait for an action submitted to the commit event loop."); + CompletableFuture completion = new CompletableFuture<>(); + runInEventLoop( + () -> { + try { + action.run(); + completion.complete(null); + } catch (Throwable t) { + completion.completeExceptionally(t); + throw t; + } }, - "handling operator event %s from subtask %d (#%d)", - event, - subtask, - attemptNumber); + actionName, + actionNameFormatParameters); + + try { + completion.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while committing batch end input", e); + } catch (ExecutionException e) { + throw new RuntimeException("Failed to commit batch end input", e.getCause()); + } } @Override 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 293b5e6b1c0b..c452a6e284ff 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 @@ -76,7 +76,8 @@ public class WriterCommittables { @VisibleForTesting WriterCommittables(CheckpointCommittables entry) { - // ? 为什么要 -1 + // 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<>(); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java index 1908ae96c26c..62d9e9a18dd3 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java @@ -170,7 +170,10 @@ private void assertCoordinatorCommitEndInput(EnvironmentSettings settings, boole .getTable(Identifier.create("default", "T_END_INPUT")); assertThat(readRowCount(table)).isEqualTo(20L); Snapshot snapshot = table.snapshotManager().latestSnapshot(); - assertThat(snapshot.watermark()).isEqualTo(endInputWatermark); + assertThat(snapshot).isNotNull(); + if (!streaming) { + assertThat(snapshot.watermark()).isEqualTo(endInputWatermark); + } } private RunningJob startStreamingInsert(boolean coordinatorCommitEnabled) throws Exception { 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 6e62e6847dc9..b078d63ae767 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 @@ -98,6 +98,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 { @@ -1345,6 +1362,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 { From 136899a12a377dd5e4a1c37fede7921b64fc893a Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Thu, 16 Jul 2026 22:00:32 +0800 Subject: [PATCH 3/8] check for TagCreationMode.BATCH --- .../java/org/apache/paimon/flink/sink/FlinkSink.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) 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 cfe20fd3d7bb..7bbd786f2983 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 @@ -52,8 +52,7 @@ import java.util.Queue; import java.util.Set; -import static org.apache.paimon.CoreOptions.WRITE_ONLY; -import static org.apache.paimon.CoreOptions.createCommitUser; +import static org.apache.paimon.CoreOptions.*; import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK; import static org.apache.paimon.flink.FlinkConnectorOptions.PRECOMMIT_COMPACT; import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT; @@ -414,6 +413,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( From 774ae9959eb551f3df70ccf1109f369ba091bd8f Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Fri, 17 Jul 2026 16:50:49 +0800 Subject: [PATCH 4/8] Pass endInputWatermark directly to coordinator to approximate original operator commit logic --- ...orCommittingRowDataStoreWriteOperator.java | 14 ++------ .../apache/paimon/flink/sink/FlinkSink.java | 17 +++++---- .../paimon/flink/sink/RowAppendTableSink.java | 24 +++++++++---- .../CommittingWriteOperatorCoordinator.java | 36 +++++++++++++++---- ...mmittingRowDataStoreWriteOperatorTest.java | 7 ++-- ...ommittingWriteOperatorCoordinatorTest.java | 12 ++++--- 6 files changed, 69 insertions(+), 41 deletions(-) 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 2b70c494fb8b..711f38a136fc 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 @@ -24,7 +24,6 @@ import org.apache.paimon.flink.sink.coordinator.CommittableEvent; import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator; import org.apache.paimon.flink.sink.coordinator.RestoredCommittableEvent; -import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.CommitMessageSerializer; import org.apache.paimon.utils.Preconditions; @@ -50,8 +49,6 @@ import java.util.NavigableMap; import java.util.TreeMap; -import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK; - /** * Write operator that hands committables to a JM-side {@link CommittingWriteOperatorCoordinator} * which performs the commit. @@ -83,8 +80,6 @@ public class CoordinatorCommittingRowDataStoreWriteOperator /** Latest watermark observed on the input; forwarded on subsequent events. */ private transient long currentWatermark; - private final Long endInputWatermark; - private transient CheckpointCommittablesSerializer stateSerializer; private transient TypeSerializer eventSerializer; @@ -96,7 +91,6 @@ public CoordinatorCommittingRowDataStoreWriteOperator( OperatorEventGateway operatorEventGateway) { super(parameters, table, storeSinkWriteProvider, initialCommitUser); this.operatorEventGateway = Preconditions.checkNotNull(operatorEventGateway); - this.endInputWatermark = new Options(table.options()).get(END_INPUT_WATERMARK); } @Override @@ -170,12 +164,8 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception { @Override protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException { List committables = prepareCommit(waitCompaction, checkpointId); - long watermark = - checkpointId == END_INPUT_CHECKPOINT_ID && endInputWatermark != null - ? endInputWatermark - : currentWatermark; CheckpointCommittables entry = - new CheckpointCommittables(checkpointId, committables, watermark); + new CheckpointCommittables(checkpointId, committables, currentWatermark); if (checkpointId == END_INPUT_CHECKPOINT_ID) { CheckpointCommittables previous = pendingCommittables.get(checkpointId); @@ -184,7 +174,7 @@ protected void emitCommittables(boolean waitCompaction, long checkpointId) throw // 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, watermark); + entry = new CheckpointCommittables(checkpointId, merged, currentWatermark); } } // Emit an event per (subtask, checkpoint) regardless of whether committables is empty. 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 7bbd786f2983..d05e56afde70 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 @@ -52,7 +52,9 @@ import java.util.Queue; import java.util.Set; -import static org.apache.paimon.CoreOptions.*; +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; import static org.apache.paimon.flink.FlinkConnectorOptions.PRECOMMIT_COMPACT; import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT; @@ -129,6 +131,7 @@ public DataStream doWrite( boolean isStreaming = isStreaming(input); boolean streamingCheckpointEnabled = isStreaming && env.getCheckpointConfig().isCheckpointingEnabled(); + Options options = Options.fromMap(table.options()); boolean writeOnly = table.coreOptions().writeOnly(); SingleOutputStreamOperator written = @@ -143,15 +146,14 @@ public DataStream doWrite( ignorePreviousFiles, hasSinkMaterializer(input)), commitUser, - streamingCheckpointEnabled)); + 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)); @@ -336,14 +338,11 @@ public static void assertBatchAdaptiveParallelism( endInputWatermark); } - /** - * Creates the writer operator factory with the runtime checkpoint mode when the writer needs - * it. Existing sinks do not use this information and keep the two-argument implementation. - */ protected OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser, - boolean streamingCheckpointEnabled) { + boolean streamingCheckpointEnabled, + @Nullable Long endInputWatermark) { return createWriteOperatorFactory(writeProvider, commitUser); } 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 9f37eec6cdfa..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,14 +50,15 @@ public RowAppendTableSink( @Override protected OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser) { - return createWriteOperatorFactory(writeProvider, commitUser, true); + return createWriteOperatorFactory(writeProvider, commitUser, true, null); } @Override protected OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser, - boolean streamingCheckpointEnabled) { + boolean streamingCheckpointEnabled, + @Nullable Long endInputWatermark) { if (coordinatorCommitEnabled()) { return createCoordinatorCommittingRowWriteOperatorFactory( table, @@ -63,7 +66,8 @@ protected OneInputStreamOperatorFactory createWriteOpe commitUser, streamingCheckpointEnabled, true, - createCommitterFactory()); + createCommitterFactory(), + endInputWatermark); } return createNoStateRowWriteOperatorFactory(table, writeProvider, commitUser); } @@ -90,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 @@ -108,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, @@ -115,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 @@ -130,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 6a819c77d7b0..453533cab2a7 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; @@ -96,18 +98,21 @@ public class CommittingWriteOperatorCoordinator implements OperatorCoordinator { 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.endInputWatermark = endInputWatermark; this.committablesSerializer = new SimpleVersionedSerializerTypeSerializerProxy<>( () -> @@ -284,7 +289,8 @@ public void notifyCheckpointComplete(long checkpointId) { boolean finalCommit = allSubtasksEndInput(); long targetCheckpointId = finalCommit ? END_INPUT_CHECKPOINT_ID : checkpointId; Map watermarkPerCheckpoint = - alignWatermarkPerCheckpoint(targetCheckpointId, subtaskCommittables); + alignWatermarkPerCheckpointForCommit( + targetCheckpointId, subtaskCommittables); commitUpToCheckpoint( targetCheckpointId, pollManifestCommittablesForCheckpoint( @@ -459,7 +465,7 @@ private void commitEndInputIfCheckpointDisabled() throws Exception { } Map watermarkPerCheckpoint = - alignWatermarkPerCheckpoint(END_INPUT_CHECKPOINT_ID, subtaskCommittables); + alignWatermarkPerCheckpointForCommit(END_INPUT_CHECKPOINT_ID, subtaskCommittables); commitUpToCheckpoint( END_INPUT_CHECKPOINT_ID, pollManifestCommittablesForCheckpoint( @@ -477,7 +483,7 @@ private void recover(long checkpointId) throws Exception { if (failoverAfterRecovery) { // recommit the restored committables and trigger a failover to reinitialize all writers Map watermarkPerCheckpoint = - alignWatermarkPerCheckpoint(checkpointId, subtaskCommittables); + alignWatermarkPerCheckpointForCommit(checkpointId, subtaskCommittables); commitUpToCheckpoint( checkpointId, pollManifestCommittablesForCheckpoint( @@ -567,6 +573,20 @@ static Map alignWatermarkPerCheckpoint( return watermarkPerCheckpoint; } + /** + * Reduces the per-subtask watermarks into the specified {@code endInputWatermark} within the + * committable when EndInput is encountered. + */ + private Map alignWatermarkPerCheckpointForCommit( + long checkpointId, WriterCommittables[] subtaskCommittables) { + Map watermarkPerCheckpoint = + alignWatermarkPerCheckpoint(checkpointId, subtaskCommittables); + if (checkpointId == END_INPUT_CHECKPOINT_ID && endInputWatermark != null) { + watermarkPerCheckpoint.put(END_INPUT_CHECKPOINT_ID, endInputWatermark); + } + return watermarkPerCheckpoint; + } + private void commitUpToCheckpoint( long checkpointId, Map toCommit, @@ -731,18 +751,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 @@ -752,7 +775,8 @@ public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) { committerFactory, streamingCheckpointEnabled, initialCommitUser, - failoverAfterRecovery); + failoverAfterRecovery, + endInputWatermark); } } } 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 f4fceadee12b..bc6bfe3c40d3 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 @@ -92,7 +92,8 @@ public void testWriterSendsCommittablesToCoordinatorAndStillEmitsDownstream() th table, table.newCommit(context.commitUser()), context), true, commitUser, - false); + false, + null); coordinator.start(); coordinator.waitProcessAllActions(); @@ -355,7 +356,7 @@ public void testEmptyRestoreStillSendsRestoreEvent() throws Exception { } @Test - public void testEndInputSendsMaxCheckpointAndUsesConfiguredWatermark() throws Exception { + public void testEndInputSendsMaxCheckpointAndForwardsCurrentWatermark() throws Exception { FileStoreTable table = createFileStoreTable( options -> { @@ -378,7 +379,7 @@ public void testEndInputSendsMaxCheckpointAndUsesConfiguredWatermark() throws Ex assertThat(event.getCheckpointId()).isEqualTo(Long.MAX_VALUE); CheckpointCommittables entry = event.deserialize(COMMITTABLES_SERIALIZER); assertThat(entry.checkpointId()).isEqualTo(Long.MAX_VALUE); - assertThat(entry.watermark()).isEqualTo(12345L); + assertThat(entry.watermark()).isEqualTo(Long.MIN_VALUE); assertThat(entry.committables()).hasSize(1); CoordinatorCommittingRowDataStoreWriteOperator operator = (CoordinatorCommittingRowDataStoreWriteOperator) harness.getOperator(); 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 b078d63ae767..cbbe8d2587da 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 @@ -295,7 +295,8 @@ public void testCheckpointDisabledCommitsWhenAllSubtasksEndInput() throws Except commitContext), false, commitUser, - false); + false, + null); coordinator.start(); coordinator.waitProcessAllActions(); @@ -533,7 +534,8 @@ public void testCheckpointFutureCompletedExceptionallyOnSnapshotFailure() throws expected), true, commitUser, - false); + false, + null); coordinator.start(); coordinator.waitProcessAllActions(); @@ -1198,7 +1200,8 @@ private CommittingWriteOperatorCoordinator createCoordinator( commitContext), true, commitUser, - failoverAfterRecovery); + failoverAfterRecovery, + null); } private CommittingWriteOperatorCoordinator createCoordinatorCapturingContext( @@ -1218,7 +1221,8 @@ private CommittingWriteOperatorCoordinator createCoordinatorCapturingContext( }, true, commitUser, - false); + false, + null); } private Committable committable(FileStoreTable table, long checkpointId, int value) From 15c1b24a35f039f30cfd650af37af03b5dd7b3c4 Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Fri, 17 Jul 2026 21:59:04 +0800 Subject: [PATCH 5/8] add EndInput ITCase --- .../paimon/flink/CoordinatorCommitITCase.java | 66 ---- .../CoordinatorEndInputCommitITCase.java | 355 ++++++++++++++++++ 2 files changed, 355 insertions(+), 66 deletions(-) create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorEndInputCommitITCase.java diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java index 62d9e9a18dd3..0ddbfbd18c48 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java @@ -18,7 +18,6 @@ package org.apache.paimon.flink; -import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.InternalRow; import org.apache.paimon.reader.RecordReader; @@ -111,71 +110,6 @@ public void testCoordinatorCommitMetricsAndCommittedRows() throws Exception { assertThat(readRowCount(runningJob.table)).isGreaterThan(0L); } - @Timeout(value = 120, unit = TimeUnit.SECONDS) - @Test - public void testCoordinatorCommitEndInputInStreamingMode() throws Exception { - assertCoordinatorCommitEndInput( - EnvironmentSettings.newInstance().inStreamingMode().build(), true); - } - - @Timeout(value = 120, unit = TimeUnit.SECONDS) - @Test - public void testCoordinatorCommitEndInputInBatchMode() throws Exception { - assertCoordinatorCommitEndInput( - EnvironmentSettings.newInstance().inBatchMode().build(), false); - } - - private void assertCoordinatorCommitEndInput(EnvironmentSettings settings, boolean streaming) - throws Exception { - final long endInputWatermark = 12345L; - TableEnvironment tEnv = TableEnvironment.create(settings); - tEnv.getConfig() - .getConfiguration() - .setString("table.exec.resource.default-parallelism", "2"); - if (streaming) { - tEnv.getConfig() - .getConfiguration() - .setString("execution.checkpointing.interval", "200 ms"); - } - - tEnv.executeSql( - "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' = '" - + tempPath - + "' )"); - tEnv.executeSql("USE CATALOG mycat"); - tEnv.executeSql( - "CREATE TABLE T_END_INPUT (id INT, data STRING) WITH (" - + "'bucket' = '-1', " - + "'write-only' = 'true', " - + "'sink.coordinator-commit.enabled' = 'true', " - + "'end-input.watermark' = '" - + endInputWatermark - + "')"); - 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')"); - - tEnv.executeSql("INSERT INTO T_END_INPUT SELECT * FROM src").await(); - - FileStoreTable table = - (FileStoreTable) - ((FlinkCatalog) tEnv.getCatalog("mycat").get()) - .catalog() - .getTable(Identifier.create("default", "T_END_INPUT")); - assertThat(readRowCount(table)).isEqualTo(20L); - Snapshot snapshot = table.snapshotManager().latestSnapshot(); - assertThat(snapshot).isNotNull(); - if (!streaming) { - assertThat(snapshot.watermark()).isEqualTo(endInputWatermark); - } - } - private RunningJob startStreamingInsert(boolean coordinatorCommitEnabled) throws Exception { String tableName = coordinatorCommitEnabled ? "T_COORDINATOR_COMMIT" : "T_DEFAULT_COMMIT"; TableEnvironment tEnv = 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..25d0a92b57df --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorEndInputCommitITCase.java @@ -0,0 +1,355 @@ +/* + * 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.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.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 testEmptyFinalCommitDoesNotUpdateEndInputWatermarkInStreamingMode() + throws Exception { + assertEndInputWatermark( + EnvironmentSettings.newInstance().inStreamingMode().build(), + true, + true, + false, + Long.MIN_VALUE); + assertEndInputWatermark( + EnvironmentSettings.newInstance().inStreamingMode().build(), + true, + false, + false, + Long.MIN_VALUE); + } + + @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 { + 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.setParallelism(DEFAULT_PARALLELISM); + streamEnv.enableCheckpointing(200L); + 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 && emitAfterFirstCheckpoint) { + // Wait for one ordinary checkpoint, then emit and finish before the next checkpoint. + // This leaves a non-empty END_INPUT committable to carry end-input.watermark. + DataStream source = + streamEnv + .fromSource( + new EmitAfterFirstCheckpointSource(), + WatermarkStrategy.noWatermarks(), + "Emit After First Checkpoint Source") + .map(row -> row) + .returns(Types.ROW(Types.INT, Types.STRING)) + .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)); + waitUntilRowCount(table, expectedRowCount); + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.watermark()).isEqualTo(expectedWatermark); + } + + /** 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; + } + } + } + + 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); + } +} From a538bf603af4e9b593fd6cc3096b10e2424bd556 Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Tue, 21 Jul 2026 21:56:08 +0800 Subject: [PATCH 6/8] [flink] Unify end-input commit completion on coordinator close --- .../CommittingWriteOperatorCoordinator.java | 62 ++++----------- .../CoordinatorEndInputCommitITCase.java | 79 +++++++++++++++++-- 2 files changed, 87 insertions(+), 54 deletions(-) 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 453533cab2a7..6d4f001e02b9 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 @@ -158,6 +158,9 @@ public void start() throws Exception { @Override public void close() throws Exception { + if (commitExecutor != null) { + drainCommitEventLoop(); + } transitionState(State.CLOSED); if (commitExecutor != null) { commitExecutor.shutdownNow(); @@ -195,11 +198,7 @@ public void checkpointCoordinator(long checkpointId, CompletableFuture r @Override public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) { - boolean waitForBatchEndInputCommit = - !streamingCheckpointEnabled - && event instanceof CommittableEvent - && ((CommittableEvent) event).getCheckpointId() == END_INPUT_CHECKPOINT_ID; - ThrowingRunnable eventHandler = + runInEventLoop( () -> { if (event instanceof CommittableEvent) { handleCommittableEvent(subtask, (CommittableEvent) event); @@ -208,60 +207,29 @@ public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEven } else { throw new UnsupportedOperationException("Unsupported event type: " + event); } - }; - - if (waitForBatchEndInputCommit) { - runInEventLoopSync( - eventHandler, - "handling operator event %s from subtask %d (#%d)", - event, - subtask, - attemptNumber); - } else { - runInEventLoop( - eventHandler, - "handling operator event %s from subtask %d (#%d)", - event, - subtask, - attemptNumber); - } + }, + "handling operator event %s from subtask %d (#%d)", + event, + subtask, + attemptNumber); } - /** - * Runs an action in the commit event loop and waits for completion. This is only used for the - * final end-input event in batch mode without checkpointing. It prevents the asynchronous - * coordinator event handling from returning before the final commit has completed, which could - * otherwise let the batch job finish prematurely. - */ - private void runInEventLoopSync( - ThrowingRunnable action, - String actionName, - Object... actionNameFormatParameters) { + /** Waits until all actions submitted before coordinator close have completed. */ + private void drainCommitEventLoop() { checkState( coordinatorThreadFactory == null || !coordinatorThreadFactory.isCurrentThreadCoordinatorThread(), - "Cannot synchronously wait for an action submitted to the commit event loop."); + "Cannot drain the commit event loop from its own thread."); CompletableFuture completion = new CompletableFuture<>(); - runInEventLoop( - () -> { - try { - action.run(); - completion.complete(null); - } catch (Throwable t) { - completion.completeExceptionally(t); - throw t; - } - }, - actionName, - actionNameFormatParameters); + runInEventLoop(() -> completion.complete(null), "draining commit event loop before close"); try { completion.get(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new RuntimeException("Interrupted while committing batch end input", e); + throw new RuntimeException("Interrupted while waiting for the commit event loop", e); } catch (ExecutionException e) { - throw new RuntimeException("Failed to commit batch end input", e.getCause()); + throw new RuntimeException("Failed to drain the commit event loop", e.getCause()); } } 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 index 25d0a92b57df..16c80bd46f8f 100644 --- 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 @@ -31,6 +31,7 @@ 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; @@ -126,13 +127,13 @@ public void testEmptyFinalCommitDoesNotUpdateEndInputWatermarkInStreamingMode() true, true, false, - Long.MIN_VALUE); + 12345L); assertEndInputWatermark( EnvironmentSettings.newInstance().inStreamingMode().build(), true, false, false, - Long.MIN_VALUE); + 12345L); } @Timeout(value = 120, unit = TimeUnit.SECONDS) @@ -195,6 +196,7 @@ private void assertEndInputWatermark( TableEnvironment tEnv; if (streaming) { streamEnv = StreamExecutionEnvironment.getExecutionEnvironment(); + streamEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING); streamEnv.setParallelism(DEFAULT_PARALLELISM); streamEnv.enableCheckpointing(200L); tEnv = StreamTableEnvironment.create(streamEnv); @@ -226,15 +228,19 @@ private void assertEndInputWatermark( + endInputWatermark + "')"); int expectedRowCount; - if (streaming && emitAfterFirstCheckpoint) { - // Wait for one ordinary checkpoint, then emit and finish before the next checkpoint. - // This leaves a non-empty END_INPUT committable to carry end-input.watermark. + 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 = streamEnv .fromSource( - new EmitAfterFirstCheckpointSource(), + emitAfterFirstCheckpoint + ? new EmitAfterFirstCheckpointSource() + : new EmitBeforeFirstCheckpointThenFinishSource(), WatermarkStrategy.noWatermarks(), - "Emit After First Checkpoint Source") + "Controlled End Input Source") .map(row -> row) .returns(Types.ROW(Types.INT, Types.STRING)) .setParallelism(1); @@ -326,6 +332,65 @@ public void notifyCheckpointComplete(long checkpointId) { } } + /** 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()); From 6a98040ba9ca1d7e411a28bad662c1dfe52793e0 Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Tue, 28 Jul 2026 09:06:43 +0800 Subject: [PATCH 7/8] add UT & codestyle --- .../apache/paimon/flink/sink/FlinkSink.java | 8 +- .../CommittingWriteOperatorCoordinator.java | 2 +- .../CoordinatorEndInputCommitITCase.java | 80 ++++++- ...mmittingRowDataStoreWriteOperatorTest.java | 22 +- ...ommittingWriteOperatorCoordinatorTest.java | 221 ++++++++++++++++++ 5 files changed, 306 insertions(+), 27 deletions(-) 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 ab6ef5cdf3a5..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 @@ -338,10 +338,10 @@ public static void assertBatchAdaptiveParallelism( } protected CommitterOperatorFactory - createCommitterOperatorFactory( - boolean streamingCheckpointEnabled, - String commitUser, - @Nullable Long endInputWatermark) { + createCommitterOperatorFactory( + boolean streamingCheckpointEnabled, + String commitUser, + @Nullable Long endInputWatermark) { return new CommitterOperatorFactory<>( streamingCheckpointEnabled, true, 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 e0fe3f3620c6..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 @@ -761,4 +761,4 @@ public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) { endInputWatermark); } } -} \ No newline at end of file +} 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 index 16c80bd46f8f..223a73e746f9 100644 --- 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 @@ -26,6 +26,7 @@ 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; @@ -118,6 +119,13 @@ public void testOperatorCommitEndInputInStreamingMode() throws Exception { 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() @@ -190,6 +198,25 @@ private void assertEndInputWatermark( 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; @@ -198,7 +225,16 @@ private void assertEndInputWatermark( streamEnv = StreamExecutionEnvironment.getExecutionEnvironment(); streamEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING); streamEnv.setParallelism(DEFAULT_PARALLELISM); - streamEnv.enableCheckpointing(200L); + 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); @@ -234,16 +270,22 @@ private void assertEndInputWatermark( // non-empty END_INPUT committable; emitting before it and ending only after it has // completed leaves an empty one. DataStream source = - 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); + 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 @@ -269,12 +311,28 @@ private void assertEndInputWatermark( ((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 { 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 70f63136369b..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 @@ -132,8 +132,8 @@ public void testWriterSendsCommittablesToCoordinatorAndStillEmitsDownstream() th @Test public void - testPendingCommittablesAccumulateAcrossUncompletedCheckpointsAndAreClearedOnComplete() - throws Exception { + testPendingCommittablesAccumulateAcrossUncompletedCheckpointsAndAreClearedOnComplete() + throws Exception { FileStoreTable table = createUnawareBucketTable(); String commitUser = UUID.randomUUID().toString(); List events = new ArrayList<>(); @@ -305,9 +305,9 @@ public void testWatermarkArrivingAfterBarrierDoesNotChangeEmittedEvent() throws harness.snapshot(2, 20); assertThat(events).hasSize(2); assertThat( - ((CommittableEvent) events.get(0)) - .deserialize(COMMITTABLES_SERIALIZER) - .watermark()) + ((CommittableEvent) events.get(0)) + .deserialize(COMMITTABLES_SERIALIZER) + .watermark()) .isEqualTo(100L); CheckpointCommittables cp2 = ((CommittableEvent) events.get(1)).deserialize(COMMITTABLES_SERIALIZER); @@ -573,11 +573,11 @@ private OneInputStreamOperatorTestHarness createHarnes new RowDataStoreWriteOperator.Factory( table, (fileStoreTable, - initialCommitUser, - state, - ioManager, - memoryPool, - metricGroup) -> + initialCommitUser, + state, + ioManager, + memoryPool, + metricGroup) -> new StoreSinkWriteImpl( fileStoreTable, initialCommitUser, @@ -662,4 +662,4 @@ public CheckpointCoordinator getCheckpointCoordinator() { return null; } } -} \ No newline at end of file +} 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 8a2185bf4bc8..3cafdc9bde30 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; @@ -310,6 +311,121 @@ public void testCheckpointDisabledCommitsWhenAllSubtasksEndInput() throws Except 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 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 { @@ -356,6 +472,105 @@ public void testRestoringAlignsBeforeRunning() throws Exception { 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 { @@ -1340,6 +1555,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(), From 09f311ce828dd04ff6d27ca3197f6cc376b027ff Mon Sep 17 00:00:00 2001 From: fishfishfishfishaa <792850842@qq.com> Date: Tue, 28 Jul 2026 17:56:05 +0800 Subject: [PATCH 8/8] add UT for JM failover situation --- ...ommittingWriteOperatorCoordinatorTest.java | 216 ++++++++++++++++++ 1 file changed, 216 insertions(+) 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 3cafdc9bde30..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 @@ -401,6 +401,62 @@ public void testPartialFailoverReplaysEndInputWhileCoordinatorIsRunning() throws 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 { @@ -472,6 +528,166 @@ 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 {