This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1851acf782 [flink] Support idle watermark handling in
coordinator-commit (#8714)
1851acf782 is described below
commit 1851acf782c4c44f19097c4d6ca7820a80e0fd70
Author: Biao Liu <[email protected]>
AuthorDate: Tue Jul 21 14:48:14 2026 +0800
[flink] Support idle watermark handling in coordinator-commit (#8714)
---
...dinatorCommittingRowDataStoreWriteOperator.java | 20 +-
.../sink/coordinator/CheckpointCommittables.java | 20 +-
.../CheckpointCommittablesSerializer.java | 12 +-
.../CommittingWriteOperatorCoordinator.java | 42 ++--
.../flink/sink/coordinator/SubtaskWatermark.java | 44 ++++
.../flink/sink/coordinator/WatermarkAligner.java | 126 ++++++++++
.../flink/sink/coordinator/WriterCommittables.java | 10 +
.../paimon/flink/CoordinatorCommitITCase.java | 254 +++++++++++++++++++++
...torCommittingRowDataStoreWriteOperatorTest.java | 90 ++++++++
.../CheckpointCommittablesSerializerTest.java | 87 +++++++
.../CommittingWriteOperatorCoordinatorTest.java | 43 +++-
.../sink/coordinator/WatermarkAlignerTest.java | 113 +++++++++
.../sink/coordinator/WriterCommittablesTest.java | 22 ++
13 files changed, 859 insertions(+), 24 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 1ff8ab06dd..3bd617b5eb 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
@@ -40,6 +40,7 @@ import
org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,6 +80,15 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
/** Latest watermark observed on the input; forwarded on subsequent
events. */
private transient long currentWatermark;
+ /**
+ * Latest {@code WatermarkStatus} observed on the input, mirroring what
Flink's upstream {@code
+ * StatusWatermarkValve} exposes. Frozen at barrier time alongside {@link
#currentWatermark} so
+ * the coordinator can reproduce valve-faithful idle handling from the
per-checkpoint entries.
+ * Not checkpointed: on restore we default to ACTIVE and let upstream
re-emit {@link
+ * WatermarkStatus#IDLE} if it still applies, matching Flink valve's
rebuilt initial state.
+ */
+ private transient boolean currentIdle;
+
private transient CheckpointCommittablesSerializer stateSerializer;
private transient TypeSerializer<CheckpointCommittables> eventSerializer;
@@ -116,6 +126,7 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
stateSerializer);
pendingCommittables = new TreeMap<>();
currentWatermark = Long.MIN_VALUE;
+ currentIdle = false;
if (context.isRestored()) {
Preconditions.checkState(context.getRestoredCheckpointId().isPresent());
@@ -159,7 +170,8 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
protected void emitCommittables(boolean waitCompaction, long checkpointId)
throws IOException {
List<Committable> committables = prepareCommit(waitCompaction,
checkpointId);
CheckpointCommittables entry =
- new CheckpointCommittables(checkpointId, committables,
currentWatermark);
+ new CheckpointCommittables(
+ checkpointId, committables, currentWatermark,
currentIdle);
// Emit an event per (subtask, checkpoint) regardless of whether
committables is empty.
operatorEventGateway.sendEventToCoordinator(
CommittableEvent.create(checkpointId, entry, eventSerializer));
@@ -182,6 +194,12 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
}
}
+ @Override
+ public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws
Exception {
+ super.processWatermarkStatus(watermarkStatus);
+ currentIdle = watermarkStatus.isIdle();
+ }
+
@VisibleForTesting
NavigableMap<Long, CheckpointCommittables> getPendingCommittables() {
return pendingCommittables;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
index 3eafc33c69..3e08c1f77f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java
@@ -32,12 +32,22 @@ public class CheckpointCommittables {
private final long checkpointId;
private final List<Committable> committables;
private final long watermark;
+ // Idle bit is frozen at barrier time together with watermark; mirrors
what Flink's
+ // StatusWatermarkValve would have observed on the writer's input at the
moment of the barrier.
+ private final boolean idle;
public CheckpointCommittables(
- long checkpointId, List<Committable> committables, long watermark)
{
+ long checkpointId, List<Committable> committables, long watermark,
boolean idle) {
this.checkpointId = checkpointId;
this.committables = committables;
this.watermark = watermark;
+ this.idle = idle;
+ }
+
+ // Convenience for callers that only need the pre-idle-aware shape (ACTIVE
writer).
+ public CheckpointCommittables(
+ long checkpointId, List<Committable> committables, long watermark)
{
+ this(checkpointId, committables, watermark, false);
}
public long checkpointId() {
@@ -52,6 +62,10 @@ public class CheckpointCommittables {
return watermark;
}
+ public boolean idle() {
+ return idle;
+ }
+
public int size() {
return committables.size();
}
@@ -63,7 +77,7 @@ public class CheckpointCommittables {
@Override
public String toString() {
return String.format(
- "CheckpointCommittables{checkpointId=%d, watermark=%d,
committables=%s}",
- checkpointId, watermark, committables);
+ "CheckpointCommittables{checkpointId=%d, watermark=%d,
idle=%s, committables=%s}",
+ checkpointId, watermark, idle, committables);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
index 9855964dff..b557559996 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java
@@ -41,7 +41,9 @@ public class CheckpointCommittablesSerializer
@Override
public int getVersion() {
- return 1;
+ // v1: checkpointId + watermark + committables
+ // v2: v1 + idle bit (appended before the committable list to keep the
ordering explicit)
+ return 2;
}
@Override
@@ -49,6 +51,7 @@ public class CheckpointCommittablesSerializer
DataOutputSerializer out = new DataOutputSerializer(256);
out.writeLong(value.checkpointId());
out.writeLong(value.watermark());
+ out.writeBoolean(value.idle());
// Nested serializer version comes before the list so the reader can
pick the right decoder
// before touching any list bytes — mirrors
ManifestCommittableSerializer's layout.
out.writeInt(committableSerializer.getVersion());
@@ -64,12 +67,15 @@ public class CheckpointCommittablesSerializer
@Override
public CheckpointCommittables deserialize(int version, byte[] serialized)
throws IOException {
- if (version != getVersion()) {
+ if (version != 1 && version != 2) {
throw new IOException("Unknown version " + version);
}
DataInputDeserializer in = new DataInputDeserializer(serialized);
long checkpointId = in.readLong();
long watermark = in.readLong();
+ // v1 payloads pre-date idle tracking; default to ACTIVE (idle=false),
which reproduces
+ // the pre-idle-aware behaviour: every subtask contributes to the min
unconditionally.
+ boolean idle = version >= 2 && in.readBoolean();
int committableVersion = in.readInt();
int count = in.readInt();
List<Committable> committables = new ArrayList<>(count);
@@ -79,6 +85,6 @@ public class CheckpointCommittablesSerializer
in.readFully(bytes);
committables.add(committableSerializer.deserialize(committableVersion, bytes));
}
- return new CheckpointCommittables(checkpointId, committables,
watermark);
+ return new CheckpointCommittables(checkpointId, committables,
watermark, idle);
}
}
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 f57e4ec149..20f3b65b88 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
@@ -83,6 +83,9 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
private final TypeSerializer<CheckpointCommittables>
committablesSerializer;
private final CoordinatorStateSerializer stateSerializer;
private final ExecutorService commitExecutor;
+ // Rebuilt per coordinator instance; state is purely in-memory, matching
Flink's
+ // StatusWatermarkValve which is also reconstructed per task instance
without checkpointing.
+ private final WatermarkAligner watermarkAligner;
// Populated by resetToCheckpoint and consumed by start. Plain fields are
sufficient: both
// callbacks run on the same scheduler thread in order.
@@ -116,6 +119,7 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
this.commitExecutor =
Executors.newSingleThreadExecutor(
new
CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context));
+ this.watermarkAligner = new WatermarkAligner(parallelism);
this.state = State.CREATED;
}
@@ -215,7 +219,8 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
throw new IllegalStateException("Not all committables
reported by writer");
}
Map<Long, Long> watermarkPerCheckpoint =
- alignWatermarkPerCheckpoint(checkpointId,
subtaskCommittables);
+ alignWatermarkPerCheckpoint(
+ checkpointId, subtaskCommittables,
watermarkAligner);
commitUpToCheckpoint(
checkpointId,
pollManifestCommittablesForCheckpoint(
@@ -352,7 +357,8 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
if (failoverAfterRecovery) {
// recommit the restored committables and trigger a failover to
reinitialize all writers
Map<Long, Long> watermarkPerCheckpoint =
- alignWatermarkPerCheckpoint(checkpointId,
subtaskCommittables);
+ alignWatermarkPerCheckpoint(
+ checkpointId, subtaskCommittables,
watermarkAligner);
commitUpToCheckpoint(
checkpointId,
pollManifestCommittablesForCheckpoint(
@@ -416,16 +422,15 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
}
/**
- * Reduce the per-subtask watermark of each checkpoint (up to {@code
checkpointId}, inclusive)
- * into a single watermark to attach to the committed snapshot. Every
subtask must have an entry
- * for {@code checkpointId} by contract (writers emit one event per
barrier, even empty), so
- * this method observes each subtask through {@link
WriterCommittables#watermarkAt}, which
- * returns {@link Long#MIN_VALUE} for missing entries — matching {@code
CommitterOperator}'s
- * initial-watermark semantics and giving idle handling a single hook to
grow into later.
+ * Aggregate each pending checkpoint's per-subtask (watermark, idle) pairs
into a single
+ * watermark by delegating to {@link WatermarkAligner}. Returns a map from
checkpoint id to the
+ * aligned watermark, covering every checkpoint up to {@code checkpointId}
inclusive.
*/
@VisibleForTesting
static Map<Long, Long> alignWatermarkPerCheckpoint(
- long checkpointId, WriterCommittables[] subtaskCommittables) {
+ long checkpointId, WriterCommittables[] subtaskCommittables,
WatermarkAligner aligner) {
+ // TreeSet keeps checkpoint ids in ascending order, matching the
aligner's contract that
+ // successive align() calls advance monotonically.
Set<Long> checkpoints = new TreeSet<>();
for (WriterCommittables committables : subtaskCommittables) {
checkpoints.addAll(
@@ -433,15 +438,24 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
}
Map<Long, Long> watermarkPerCheckpoint = new HashMap<>();
for (long cp : checkpoints) {
- long min = Long.MAX_VALUE;
- for (WriterCommittables committables : subtaskCommittables) {
- min = Math.min(min, committables.watermarkAt(cp));
- }
- watermarkPerCheckpoint.put(cp, min);
+ watermarkPerCheckpoint.put(
+ cp, aligner.align(subtaskWatermarksAt(cp,
subtaskCommittables)));
}
return watermarkPerCheckpoint;
}
+ private static SubtaskWatermark[] subtaskWatermarksAt(
+ long checkpointId, WriterCommittables[] subtaskCommittables) {
+ SubtaskWatermark[] subtaskWatermarks = new
SubtaskWatermark[subtaskCommittables.length];
+ for (int i = 0; i < subtaskCommittables.length; i++) {
+ subtaskWatermarks[i] =
+ new SubtaskWatermark(
+ subtaskCommittables[i].watermarkAt(checkpointId),
+ subtaskCommittables[i].isIdleAt(checkpointId));
+ }
+ return subtaskWatermarks;
+ }
+
private void commitUpToCheckpoint(
long checkpointId,
Map<Long, ManifestCommittable> toCommit,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SubtaskWatermark.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SubtaskWatermark.java
new file mode 100644
index 0000000000..ef83aacfb2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SubtaskWatermark.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+/**
+ * Point-in-time watermark reading from a single subtask, paired with whether
that subtask was idle
+ * at the moment the reading was taken. Consumed by {@link WatermarkAligner}
to reproduce {@code
+ * StatusWatermarkValve} semantics over inputs that arrive via the {@code
OperatorEvent} channel
+ * rather than the operator's own edge.
+ */
+public final class SubtaskWatermark {
+
+ private final long watermark;
+ private final boolean idle;
+
+ public SubtaskWatermark(long watermark, boolean idle) {
+ this.watermark = watermark;
+ this.idle = idle;
+ }
+
+ public long watermark() {
+ return watermark;
+ }
+
+ public boolean idle() {
+ return idle;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkAligner.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkAligner.java
new file mode 100644
index 0000000000..277b7503c2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkAligner.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+/**
+ * Coordinator-side counterpart of Flink's {@code StatusWatermarkValve}.
Aggregates a per-subtask
+ * ({@link SubtaskWatermark} array indexed by subtask id) into a single
monotonically non-decreasing
+ * watermark.
+ *
+ * <p>The algorithm mirrors {@code StatusWatermarkValve} channel-by-channel:
+ *
+ * <ul>
+ * <li>A subtask that reports {@code idle=true} becomes unaligned and is
excluded from the
+ * aligned-set min.
+ * <li>A subtask that reports {@code idle=false} rejoins the aligned set
once its watermark has
+ * caught up to the last emitted watermark; while lagging it stays
unaligned so it does not
+ * drag the global watermark backwards.
+ * <li>If every subtask is unaligned (all idle, or all lagging), the aligner
emits a one-shot
+ * "flush max" — the maximum over every subtask's last known watermark —
matching what {@code
+ * StatusWatermarkValve} does when the last active channel transitions
to IDLE.
+ * </ul>
+ *
+ * <p>State is purely in-memory: the aligner mirrors {@code
StatusWatermarkValve}'s "no checkpoint,
+ * rebuild on restart" contract. A fresh instance starts every subtask ACTIVE
with {@code
+ * Long.MIN_VALUE}, so upstream must re-send {@code WatermarkStatus.IDLE} for
it to take effect
+ * again — same as Flink's built-in valve on task restart.
+ */
+public class WatermarkAligner {
+
+ private final boolean[] aligned;
+
+ private long lastEmittedWatermark;
+ private boolean idleStatus;
+
+ public WatermarkAligner(int parallelism) {
+ this.aligned = new boolean[parallelism];
+ for (int i = 0; i < parallelism; i++) {
+ aligned[i] = true;
+ }
+ this.lastEmittedWatermark = Long.MIN_VALUE;
+ this.idleStatus = false;
+ }
+
+ /**
+ * Aggregate the given per-subtask readings into a single watermark.
Updates the aligner's
+ * internal per-subtask alignment as a side effect, so successive calls
see valve-faithful
+ * catch-up semantics.
+ *
+ * <p>Contract: successive calls must correspond to strictly increasing
checkpoint ids. Calling
+ * out of order would apply a later checkpoint's alignment side effects to
an earlier one,
+ * silently corrupting the emitted watermark sequence.
+ *
+ * @return the aligned watermark for this call, guaranteed to be
monotonically non-decreasing
+ * across successive calls.
+ */
+ public long align(SubtaskWatermark[] subtaskWatermarks) {
+ int parallelism = subtaskWatermarks.length;
+ if (parallelism != aligned.length) {
+ throw new IllegalStateException(
+ "Aligner parallelism "
+ + aligned.length
+ + " does not match input "
+ + parallelism);
+ }
+
+ long alignedMin = Long.MAX_VALUE;
+ boolean anyAligned = false;
+ for (int i = 0; i < parallelism; i++) {
+ long watermark = subtaskWatermarks[i].watermark();
+ boolean idle = subtaskWatermarks[i].idle();
+ if (idle) {
+ aligned[i] = false;
+ continue;
+ }
+ if (!aligned[i] && watermark >= lastEmittedWatermark) {
+ aligned[i] = true;
+ }
+ if (aligned[i]) {
+ anyAligned = true;
+ if (watermark < alignedMin) {
+ alignedMin = watermark;
+ }
+ }
+ }
+
+ if (anyAligned) {
+ if (alignedMin > lastEmittedWatermark) {
+ lastEmittedWatermark = alignedMin;
+ }
+ idleStatus = false;
+ } else if (!idleStatus) {
+ // All-unaligned transition: flush the max across every subtask's
last known watermark
+ // (equivalent to
StatusWatermarkValve.findAndOutputMaxWatermarkAcrossAllSubpartitions),
+ // then latch to IDLE so subsequent all-unaligned checkpoints do
not re-flush.
+ long flushMax = Long.MIN_VALUE;
+ for (int i = 0; i < parallelism; i++) {
+ long watermark = subtaskWatermarks[i].watermark();
+ if (watermark > flushMax) {
+ flushMax = watermark;
+ }
+ }
+ if (flushMax > lastEmittedWatermark) {
+ lastEmittedWatermark = flushMax;
+ }
+ idleStatus = true;
+ }
+
+ return lastEmittedWatermark;
+ }
+}
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 243969069b..1562fbf4ee 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
@@ -135,6 +135,16 @@ public class WriterCommittables {
return entry == null ? Long.MIN_VALUE : entry.watermark();
}
+ /**
+ * Returns whether this subtask was idle at {@code checkpointId}. Falls
back to {@code false}
+ * (ACTIVE) when the subtask has no entry — equivalent to "channel exists
but has never reported
+ * a watermark yet", which Flink's {@code StatusWatermarkValve} also
treats as ACTIVE.
+ */
+ public boolean isIdleAt(long checkpointId) {
+ CheckpointCommittables entry =
committablesPerCheckpoint.get(checkpointId);
+ return entry != null && entry.idle();
+ }
+
@Override
public String toString() {
return String.format(
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 0ddbfbd18c..eaa541e099 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,26 +18,42 @@
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.flink.sink.FlinkSinkBuilder;
+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.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.JobID;
+import org.apache.flink.api.common.eventtime.Watermark;
+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.execution.JobClient;
+import org.apache.flink.core.io.InputStatus;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.OperatorMetricGroup;
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.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -54,6 +70,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class CoordinatorCommitITCase {
private static final int DEFAULT_PARALLELISM = 2;
+ private static final int SCRIPTED_PARALLELISM = 2;
private static final long WAIT_TIMEOUT_MILLIS = 60_000L;
private static final InMemoryReporter reporter = InMemoryReporter.create();
@@ -110,6 +127,114 @@ public class CoordinatorCommitITCase {
assertThat(readRowCount(runningJob.table)).isGreaterThan(0L);
}
+ /**
+ * Idle watermark parity: the snapshot watermark observed with
coordinator-commit enabled must
+ * match the one produced by the classic {@code CommitterOperator} path
under the same input
+ * script. See {@link IdleWatermarkScriptedSource} for the three scenarios
covered — steady
+ * multi-active min, one subtask idle, and all subtasks idle.
+ */
+ @Timeout(value = 240, unit = TimeUnit.SECONDS)
+ @Test
+ public void testIdleWatermarkParityAcrossCommitPaths() throws Exception {
+ for (IdleWatermarkScriptedSource.Scenario scenario :
+ IdleWatermarkScriptedSource.Scenario.values()) {
+ long coordinatorWatermark = runIdleWatermarkScenario(scenario,
true);
+ long committerWatermark = runIdleWatermarkScenario(scenario,
false);
+ assertThat(coordinatorWatermark)
+ .describedAs(
+ "coordinator vs committer snapshot watermark
mismatch for scenario "
+ + scenario)
+ .isEqualTo(committerWatermark);
+ assertThat(coordinatorWatermark)
+ .describedAs("scenario " + scenario + " expected
watermark")
+ .isEqualTo(scenario.expectedWatermark);
+ }
+ }
+
+ private long runIdleWatermarkScenario(
+ IdleWatermarkScriptedSource.Scenario scenario, boolean
coordinatorCommitEnabled)
+ throws Exception {
+ String tableName =
+ "T_IDLE_"
+ + scenario.name()
+ + "_"
+ + (coordinatorCommitEnabled ? "COORD" : "CLASSIC");
+ TableEnvironment tEnv =
+ TableEnvironment.create(
+
EnvironmentSettings.newInstance().inStreamingMode().build());
+
tEnv.getConfig().getConfiguration().setString("execution.checkpointing.interval",
"200 ms");
+ tEnv.executeSql(
+ "CREATE CATALOG idlecat WITH ( 'type' = 'paimon', 'warehouse'
= '"
+ + tempPath
+ + "/"
+ + scenario.name()
+ + '_'
+ + coordinatorCommitEnabled
+ + "' )");
+ tEnv.executeSql("USE CATALOG idlecat");
+ // write-only=true is kept identical across both paths so the only
variable under test is
+ // sink.coordinator-commit.enabled; otherwise compaction snapshots
would perturb the
+ // "latest snapshot watermark" observations differently between paths.
+ String coordinatorOption =
+ coordinatorCommitEnabled ? ",
'sink.coordinator-commit.enabled' = 'true'" : "";
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tableName
+ + " (id INT, data STRING) WITH ("
+ + "'bucket' = '-1', 'write-only' = 'true'"
+ + coordinatorOption
+ + ")");
+
+ FileStoreTable table =
+ (FileStoreTable)
+ ((FlinkCatalog) tEnv.getCatalog("idlecat").get())
+ .catalog()
+ .getTable(Identifier.create("default",
tableName));
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(SCRIPTED_PARALLELISM);
+ env.enableCheckpointing(200);
+
+ DataStreamSource<RowData> source =
+ env.fromSource(
+ new IdleWatermarkScriptedSource(scenario),
+
org.apache.flink.api.common.eventtime.WatermarkStrategy
+ .noWatermarks(),
+ "idle-watermark-source")
+ .setParallelism(SCRIPTED_PARALLELISM);
+
+ new FlinkSinkBuilder(table).forRowData(source).build();
+
+ JobClient client = env.executeAsync("idle-watermark-" + tableName);
+ JobID jobId = client.getJobID();
+ try {
+ // The option only expresses intent; a failed precondition would
silently fall back to
+ // the classic path. Confirm the intended path is actually running
by waiting for its
+ // own positive metric signal before trusting any watermark
observation.
+ if (coordinatorCommitEnabled) {
+ waitUntilCoordinatorCommitMetricsRegistered(jobId);
+ } else {
+ waitUntilGlobalCommitterMetricGroupsRegistered(jobId);
+ }
+
+ long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS;
+ long observedWatermark = Long.MIN_VALUE;
+ while (System.currentTimeMillis() < deadline) {
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ if (snapshot != null && snapshot.watermark() != null) {
+ observedWatermark = snapshot.watermark();
+ if (observedWatermark >= scenario.expectedWatermark) {
+ break;
+ }
+ }
+ Thread.sleep(200);
+ }
+ return observedWatermark;
+ } finally {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ }
+ }
+
private RunningJob startStreamingInsert(boolean coordinatorCommitEnabled)
throws Exception {
String tableName = coordinatorCommitEnabled ? "T_COORDINATOR_COMMIT" :
"T_DEFAULT_COMMIT";
TableEnvironment tEnv =
@@ -303,4 +428,133 @@ public class CoordinatorCommitITCase {
client.cancel().get(30, TimeUnit.SECONDS);
}
}
+
+ /**
+ * Scripted source that emits controlled ({@link Watermark}, {@link
+ * org.apache.flink.api.connector.source.SourceOutput#markIdle()})
sequences on a per-subtask
+ * basis, so the same input drives both commit paths deterministically
without depending on
+ * datagen timing. Subtask-0 and subtask-1 follow the {@code Scenario}'s
script; each step emits
+ * a watermark or marks the split idle, sleeps briefly to let checkpoints
run, then the source
+ * stays available forever so the job is only stopped by the outer cancel.
+ */
+ private static class IdleWatermarkScriptedSource extends
AbstractNonCoordinatedSource<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ enum Scenario {
+ /** Both subtasks stay active; snapshot watermark equals the
smaller of the two. */
+ ALL_ACTIVE(300L),
+ /**
+ * Subtask-0 emits a small watermark then goes idle; subtask-1
keeps emitting bigger
+ * watermarks. The idle subtask must not hold the snapshot back.
+ */
+ PARTIAL_IDLE(700L),
+ /**
+ * Both subtasks emit a watermark and go idle. Once every input is
idle, Flink's valve
+ * flushes {@code max} over all channels — so the snapshot
watermark advances to the
+ * larger of the two, then stays put (not regresses back to {@code
Long.MIN_VALUE}).
+ */
+ ALL_IDLE(600L);
+
+ final long expectedWatermark;
+
+ Scenario(long expectedWatermark) {
+ this.expectedWatermark = expectedWatermark;
+ }
+ }
+
+ private final Scenario scenario;
+
+ IdleWatermarkScriptedSource(Scenario scenario) {
+ this.scenario = scenario;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<RowData, SimpleSourceSplit> createReader(
+ SourceReaderContext sourceReaderContext) {
+ return new Reader(scenario,
sourceReaderContext.getIndexOfSubtask());
+ }
+
+ private static class Reader extends
AbstractNonCoordinatedSourceReader<RowData> {
+
+ private final Scenario scenario;
+ private final int subtaskIndex;
+
+ private int step;
+
+ Reader(Scenario scenario, int subtaskIndex) {
+ this.scenario = scenario;
+ this.subtaskIndex = subtaskIndex;
+ }
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<RowData> output) throws
InterruptedException {
+ if (step == 0) {
+ // Emit one record so the writer has something to commit
and thus produces a
+ // snapshot carrying the watermark for us to observe.
+ output.collect(
+ GenericRowData.of(
+ subtaskIndex, StringData.fromString("s" +
subtaskIndex)));
+ }
+ switch (scenario) {
+ case ALL_ACTIVE:
+ return driveAllActive(output);
+ case PARTIAL_IDLE:
+ return drivePartialIdle(output);
+ case ALL_IDLE:
+ return driveAllIdle(output);
+ default:
+ throw new IllegalStateException("Unknown scenario " +
scenario);
+ }
+ }
+
+ private InputStatus driveAllActive(ReaderOutput<RowData> output)
+ throws InterruptedException {
+ // Both subtasks keep emitting increasing watermarks. Min
across subtasks = 300.
+ long watermark = subtaskIndex == 0 ? 300L : 500L;
+ output.emitWatermark(new Watermark(watermark));
+ Thread.sleep(200);
+ step++;
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ private InputStatus drivePartialIdle(ReaderOutput<RowData> output)
+ throws InterruptedException {
+ if (subtaskIndex == 0) {
+ if (step == 0) {
+ output.emitWatermark(new Watermark(100L));
+ Thread.sleep(200);
+ output.markIdle();
+ }
+ Thread.sleep(200);
+ step++;
+ return InputStatus.MORE_AVAILABLE;
+ }
+ // subtask-1 keeps advancing; peers at 700 by the second step.
+ long watermark = 500L + Math.min(step, 1) * 200L;
+ output.emitWatermark(new Watermark(watermark));
+ Thread.sleep(200);
+ step++;
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ private InputStatus driveAllIdle(ReaderOutput<RowData> output)
+ throws InterruptedException {
+ if (step == 0) {
+ long watermark = subtaskIndex == 0 ? 400L : 600L;
+ output.emitWatermark(new Watermark(watermark));
+ Thread.sleep(200);
+ output.markIdle();
+ }
+ Thread.sleep(200);
+ step++;
+ return InputStatus.MORE_AVAILABLE;
+ }
+ }
+ }
}
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 296eaa3823..bae18ea2c2 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
@@ -48,6 +48,7 @@ import
org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.FlinkRuntimeException;
import org.junit.jupiter.api.Test;
@@ -353,6 +354,95 @@ public class
CoordinatorCommittingRowDataStoreWriteOperatorTest extends Committe
secondHarness.close();
}
+ @Test
+ @Timeout(30)
+ public void testWatermarkStatusFrozenAtBarrierAcrossCheckpoints() throws
Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ List<OperatorEvent> events = new ArrayList<>();
+
+ OneInputStreamOperatorTestHarness<InternalRow, Committable> harness =
+ createHarness(table, commitUser, events::add);
+ TypeSerializer<Committable> committableSerializer =
+ new CommittableTypeInfo().createSerializer(new
ExecutionConfig());
+ harness.setup(committableSerializer);
+ harness.open();
+
+ // cp1: writer is IDLE at barrier time.
+ harness.processWatermark(new Watermark(100L));
+ harness.processWatermarkStatus(WatermarkStatus.IDLE);
+ harness.prepareSnapshotPreBarrier(1);
+ harness.snapshot(1, 10);
+
+ // cp2: back to ACTIVE with a new watermark. Idle status must not
linger on cp2 just
+ // because cp1 was idle.
+ harness.processWatermarkStatus(WatermarkStatus.ACTIVE);
+ harness.processWatermark(new Watermark(500L));
+ harness.prepareSnapshotPreBarrier(2);
+ harness.snapshot(2, 20);
+
+ assertThat(events).hasSize(2);
+ CheckpointCommittables cp1 =
+ ((CommittableEvent)
events.get(0)).deserialize(COMMITTABLES_SERIALIZER);
+ assertThat(cp1.checkpointId()).isEqualTo(1L);
+ assertThat(cp1.watermark()).isEqualTo(100L);
+ assertThat(cp1.idle()).isTrue();
+
+ CheckpointCommittables cp2 =
+ ((CommittableEvent)
events.get(1)).deserialize(COMMITTABLES_SERIALIZER);
+ assertThat(cp2.checkpointId()).isEqualTo(2L);
+ assertThat(cp2.watermark()).isEqualTo(500L);
+ assertThat(cp2.idle()).isFalse();
+
+ harness.close();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testIdleFlagResetsToActiveOnRestore() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ String commitUser = UUID.randomUUID().toString();
+ TypeSerializer<Committable> committableSerializer =
+ new CommittableTypeInfo().createSerializer(new
ExecutionConfig());
+
+ // session 1: end the session while IDLE so the restore path is
exercised on a snapshot
+ // taken from an idle writer.
+ List<OperatorEvent> firstEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness<InternalRow, Committable>
firstHarness =
+ createHarness(table, commitUser, firstEvents::add);
+ firstHarness.setup(committableSerializer);
+ firstHarness.open();
+
+ firstHarness.processWatermark(new Watermark(100L));
+ firstHarness.processWatermarkStatus(WatermarkStatus.IDLE);
+ firstHarness.prepareSnapshotPreBarrier(1);
+ OperatorSubtaskState snapshot = firstHarness.snapshot(1, 10);
+ firstHarness.close();
+
+ // session 2: restore. Flink runtime does not replay the last
WatermarkStatus, and the
+ // aligner treats channels as ACTIVE + Long.MIN_VALUE on rebuild, so
the writer must also
+ // default to ACTIVE. The next barrier — before any WatermarkStatus
event — must emit
+ // idle=false, matching Flink's own valve-rebuild contract.
+ List<OperatorEvent> restoredEvents = new ArrayList<>();
+ OneInputStreamOperatorTestHarness<InternalRow, Committable>
secondHarness =
+ createHarness(table, commitUser, restoredEvents::add);
+ secondHarness.setup(committableSerializer);
+ restoreWithCheckpointId(secondHarness, snapshot, 1L);
+ secondHarness.open();
+
+ secondHarness.prepareSnapshotPreBarrier(2);
+ secondHarness.snapshot(2, 20);
+
+ // First event is the restore replay; the second is the freshly frozen
cp2.
+ assertThat(restoredEvents).hasSize(2);
+ CommittableEvent cp2Event = (CommittableEvent) restoredEvents.get(1);
+ CheckpointCommittables cp2 =
cp2Event.deserialize(COMMITTABLES_SERIALIZER);
+ assertThat(cp2.checkpointId()).isEqualTo(2L);
+ assertThat(cp2.idle()).isFalse();
+
+ secondHarness.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/coordinator/CheckpointCommittablesSerializerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
new file mode 100644
index 0000000000..09b894d03d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializerTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.apache.paimon.flink.sink.CommittableSerializer;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link CheckpointCommittablesSerializer}. */
+public class CheckpointCommittablesSerializerTest {
+
+ private final CheckpointCommittablesSerializer serializer =
+ new CheckpointCommittablesSerializer(
+ new CommittableSerializer(new CommitMessageSerializer()));
+
+ @Test
+ public void testCurrentVersionIsV2() {
+ assertThat(serializer.getVersion()).isEqualTo(2);
+ }
+
+ @Test
+ public void testRoundTripPreservesIdleFlag() throws IOException {
+ for (boolean idle : new boolean[] {true, false}) {
+ CheckpointCommittables original =
+ new CheckpointCommittables(
+ 42L, Collections.emptyList(), /* watermark */
4242L, idle);
+ CheckpointCommittables decoded =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(original));
+ assertThat(decoded.checkpointId()).isEqualTo(42L);
+ assertThat(decoded.watermark()).isEqualTo(4242L);
+ assertThat(decoded.idle()).isEqualTo(idle);
+ assertThat(decoded.committables()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testV1PayloadDeserializesAsActive() throws IOException {
+ // Hand-encode a v1 payload (no idle bit) so the reader is exercised
on real bytes rather
+ // than a spec that could drift alongside the serializer
implementation.
+ DataOutputSerializer out = new DataOutputSerializer(32);
+ out.writeLong(7L); // checkpointId
+ out.writeLong(1234L); // watermark
+ out.writeInt(new CommittableSerializer(new
CommitMessageSerializer()).getVersion());
+ out.writeInt(0); // empty committables
+
+ CheckpointCommittables decoded = serializer.deserialize(1,
out.getCopyOfBuffer());
+ assertThat(decoded.checkpointId()).isEqualTo(7L);
+ assertThat(decoded.watermark()).isEqualTo(1234L);
+ // v1 predates idle tracking; readers must default to ACTIVE so
pre-upgrade payloads keep
+ // participating in the min just like they did before.
+ assertThat(decoded.idle()).isFalse();
+ assertThat(decoded.committables()).isEmpty();
+ }
+
+ @Test
+ public void testUnknownVersionRejected() throws IOException {
+ byte[] bytes =
+ serializer.serialize(new CheckpointCommittables(1L,
Collections.emptyList(), 0L));
+ assertThatThrownBy(() -> serializer.deserialize(99, bytes))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unknown version");
+ }
+}
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 5889f41cf5..a43bba18d3 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
@@ -552,6 +552,33 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
coordinator.close();
}
+ /**
+ * When one subtask marks itself idle, the coordinator must skip that
subtask and take the min
+ * only over the remaining active subtasks — mirroring Flink's {@code
StatusWatermarkValve}.
+ */
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testAlignmentSkipsIdleSubtaskWhenSomeActive() throws Exception
{
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 2);
+ CommittingWriteOperatorCoordinator coordinator =
createCoordinator(table, context, false);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ // cp1: subtask-0 IDLE at watermark=100, subtask-1 ACTIVE with data +
watermark=500.
+ // Without idle handling this would emit min=100 and hold the snapshot
back to 100.
+ // With idle handling subtask-0 is excluded and the snapshot watermark
advances to 500.
+ coordinator.handleEventFromOperator(0, 0, idleEvent(1L, 100L));
+ coordinator.handleEventFromOperator(1, 0, event(500L,
committable(table, 1, 1)));
+ coordinator.notifyCheckpointComplete(1L);
+ coordinator.waitProcessAllActions();
+
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(snapshot).isNotNull();
+ assertThat(snapshot.watermark()).isEqualTo(500L);
+ coordinator.close();
+ }
+
@Timeout(value = 30, unit = TimeUnit.SECONDS)
@Test
public void testPollManifestCommittablesForCheckpoint() throws Exception {
@@ -662,7 +689,9 @@ public class CommittingWriteOperatorCoordinatorTest extends
CommitterOperatorTes
checkpointId2,
writerCommittables,
CommittingWriteOperatorCoordinator.alignWatermarkPerCheckpoint(
- checkpointId2, writerCommittables),
+ checkpointId2,
+ writerCommittables,
+ new
WatermarkAligner(writerCommittables.length)),
committer);
BinaryRow partition = new BinaryRow(1);
@@ -751,13 +780,13 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
Map<Long, Long> upToCp1 =
CommittingWriteOperatorCoordinator.alignWatermarkPerCheckpoint(
- 1L, writerCommittables);
+ 1L, writerCommittables, new
WatermarkAligner(writerCommittables.length));
assertThat(upToCp1).hasSize(1);
assertThat(upToCp1.get(1L)).isEqualTo(100L);
Map<Long, Long> upToCp2 =
CommittingWriteOperatorCoordinator.alignWatermarkPerCheckpoint(
- 2L, writerCommittables);
+ 2L, writerCommittables, new
WatermarkAligner(writerCommittables.length));
assertThat(upToCp2).hasSize(2);
assertThat(upToCp2.get(1L)).isEqualTo(100L);
assertThat(upToCp2.get(2L)).isEqualTo(400L);
@@ -1180,6 +1209,14 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterOperatorTes
SERIALIZER);
}
+ private CommittableEvent idleEvent(long checkpointId, long watermark)
throws Exception {
+ return CommittableEvent.create(
+ checkpointId,
+ new CheckpointCommittables(
+ checkpointId, Collections.emptyList(), watermark, /*
idle */ true),
+ SERIALIZER);
+ }
+
private CommittableEvent emptyEvent(long checkpointId) throws Exception {
return eventOf(checkpointId, Collections.emptyList(), Long.MIN_VALUE);
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkAlignerTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkAlignerTest.java
new file mode 100644
index 0000000000..7aec09a48f
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkAlignerTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link WatermarkAligner}. */
+public class WatermarkAlignerTest {
+
+ private static final int PARALLELISM = 3;
+
+ @Test
+ public void testAllActiveTakesMinAcrossSubtasks() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ assertThat(aligner.align(new SubtaskWatermark[] {active(100L),
active(200L), active(300L)}))
+ .isEqualTo(100L);
+ }
+
+ @Test
+ public void testMonotonicMinAcrossCheckpoints() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ aligner.align(new SubtaskWatermark[] {active(100L), active(200L),
active(300L)});
+ // Min across subtasks for the next reading is 250 — greater than the
last emit of 100.
+ assertThat(aligner.align(new SubtaskWatermark[] {active(400L),
active(250L), active(500L)}))
+ .isEqualTo(250L);
+ }
+
+ @Test
+ public void testIdleSubtaskIsSkippedAndActiveMinAdvances() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ aligner.align(new SubtaskWatermark[] {active(100L), active(200L),
active(300L)});
+ // subtask-0 goes IDLE; remaining actives push min to 400.
+ assertThat(aligner.align(new SubtaskWatermark[] {idle(100L),
active(400L), active(500L)}))
+ .isEqualTo(400L);
+ }
+
+ @Test
+ public void testAllIdleFlushesMaxThenLatches() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ aligner.align(new SubtaskWatermark[] {active(100L), active(200L),
active(300L)});
+
+ // Everyone goes IDLE with their last known watermarks — flush max=300.
+ assertThat(aligner.align(new SubtaskWatermark[] {idle(100L),
idle(200L), idle(300L)}))
+ .isEqualTo(300L);
+
+ // Still all idle — flush must not re-fire, emission stays at 300.
+ assertThat(aligner.align(new SubtaskWatermark[] {idle(100L),
idle(200L), idle(350L)}))
+ .isEqualTo(300L);
+ }
+
+ @Test
+ public void testIdleThenActiveWithLagStaysHeldUntilCatchUp() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ aligner.align(new SubtaskWatermark[] {active(100L), active(200L),
active(300L)});
+
+ // subtask-0 IDLE; actives advance min to 250.
+ assertThat(aligner.align(new SubtaskWatermark[] {idle(100L),
active(250L), active(400L)}))
+ .isEqualTo(250L);
+
+ // subtask-0 comes back ACTIVE at 150 — still lagging (< last=250) so
it stays unaligned;
+ // min over aligned actives is min(300, 500) = 300.
+ assertThat(aligner.align(new SubtaskWatermark[] {active(150L),
active(300L), active(500L)}))
+ .isEqualTo(300L);
+
+ // subtask-0 climbs to 260 (< last=300) — still lagging, aligned min =
320.
+ assertThat(aligner.align(new SubtaskWatermark[] {active(260L),
active(320L), active(600L)}))
+ .isEqualTo(320L);
+
+ // subtask-0 finally reaches 350 >= last=320, rejoins the aligned set;
new min=350.
+ assertThat(aligner.align(new SubtaskWatermark[] {active(350L),
active(400L), active(700L)}))
+ .isEqualTo(350L);
+ }
+
+ @Test
+ public void testAbsentSubtaskBlocksMinLikeNeverReported() {
+ WatermarkAligner aligner = new WatermarkAligner(PARALLELISM);
+ // Absent subtasks are represented by ACTIVE + Long.MIN_VALUE — the
same shape Flink's
+ // valve exposes for a channel that has never emitted a watermark.
Under that reading the
+ // min stays at Long.MIN_VALUE (never-reported subtasks block the min,
unlike idle ones).
+ assertThat(
+ aligner.align(
+ new SubtaskWatermark[] {
+ active(500L), active(Long.MIN_VALUE),
active(Long.MIN_VALUE)
+ }))
+ .isEqualTo(Long.MIN_VALUE);
+ }
+
+ private static SubtaskWatermark active(long watermark) {
+ return new SubtaskWatermark(watermark, false);
+ }
+
+ private static SubtaskWatermark idle(long watermark) {
+ return new SubtaskWatermark(watermark, true);
+ }
+}
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 74352dbed4..aa2b20c832 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
@@ -330,6 +330,28 @@ public class WriterCommittablesTest {
Long.MIN_VALUE))));
}
+ @Test
+ public void testIsIdleAtDistinguishesPresentAndAbsent() {
+ WriterCommittables committables =
+ new WriterCommittables(
+ new CheckpointCommittables(
+ 1L, Collections.emptyList(), 100L, /* idle */
true));
+ committables.mergeWith(
+ new WriterCommittables(
+ new CheckpointCommittables(
+ 2L, Collections.emptyList(), 200L, /* idle */
false)));
+
+ // present + idle=true
+ assertThat(committables.isIdleAt(1L)).isTrue();
+ assertThat(committables.watermarkAt(1L)).isEqualTo(100L);
+ // present + idle=false
+ assertThat(committables.isIdleAt(2L)).isFalse();
+ assertThat(committables.watermarkAt(2L)).isEqualTo(200L);
+ // absent: mirrors Flink valve's "channel exists but never reported" —
ACTIVE with MIN_VALUE
+ assertThat(committables.isIdleAt(3L)).isFalse();
+ assertThat(committables.watermarkAt(3L)).isEqualTo(Long.MIN_VALUE);
+ }
+
@Test
public void testDuplicateCheckpointIdRejected() {
CommitMessage commitMessage = createEmptyCommitMessage();
