From 68bc0b0a0d51b79789263e8949da1115624d7e1c Mon Sep 17 00:00:00 2001 From: developer-rpai Date: Fri, 9 Oct 2026 16:05:45 -0700 Subject: [PATCH 1/8] Flink: Forward watermarks in EqualityConvertCommitter when no cycle is active Fixes #18435. --- .../operator/EqualityConvertCommitter.java | 10 +++++++++- .../operator/TestEqualityConvertCommitter.java | 18 ++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index ada16ec09384..5c080ac64c9a 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -148,7 +148,15 @@ public void processElement2(StreamRecord record) { @Override public void processWatermark(Watermark mark) throws Exception { - if (planResult == null || mark.getTimestamp() < planResult.doneTimestamp()) { + if (planResult == null) { + // No conversion cycle is active: forward the watermark. It may originate from another + // task's trigger sharing the maintenance lock, and holding it back would stall lock + // release for that task. + super.processWatermark(mark); + return; + } + + if (mark.getTimestamp() < planResult.doneTimestamp()) { // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the // lock early and could let the next trigger run a concurrent cycle on the same staging diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 53e98a06b478..1bca49a39fc8 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -133,6 +133,24 @@ void holdsBackWatermarkUntilCommit() throws Exception { } } + @Test + void forwardsWatermarkWithoutActiveCycle() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + insert(table, 1, "a"); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + // No plan was received, so no conversion cycle is active. A watermark here originates + // from another task's trigger sharing the maintenance lock; dropping it would stall + // the unioned watermark and the lock would never be released. + long markTs = System.currentTimeMillis(); + harness.processBothWatermarks(new Watermark(markTs)); + assertThat(watermarks(harness)).containsExactly(new Watermark(markTs)); + } + } + @Test void skipsCommitForEmptyCycle() throws Exception { Table table = createTable(3, FileFormat.PARQUET); From bf9047043967b85b24149e43cbc7b837199950a6 Mon Sep 17 00:00:00 2001 From: Rakesh Pai <41351936+developer-rpai@users.noreply.github.com> Date: Sun, 11 Oct 2026 20:50:33 -0700 Subject: [PATCH 2/8] Flink: Gate EqualityConvertCommitter watermarks on cycle-start marker The planner emits phase watermarks before the plan reaches the committer. Forwarding watermarks while no plan has been received would release the maintenance lock early, reintroducing the race fixed in #17038. The planner now emits a cycle-start marker on the metadata stream before its first phase watermark; the committer forwards watermarks only while no cycle is active and holds them back from the marker until the commit. --- .../operator/EqualityConvertCommitter.java | 42 ++++++++++++------ .../operator/EqualityConvertDVWriter.java | 4 ++ .../operator/EqualityConvertPlan.java | 27 ++++++++++++ .../operator/EqualityConvertPlanner.java | 10 ++++- .../TestEqualityConvertCommitter.java | 43 +++++++++++++++++++ .../operator/TestEqualityConvertDVWriter.java | 30 +++++++++++++ .../operator/TestEqualityConvertPlanner.java | 19 ++++---- 7 files changed, 154 insertions(+), 21 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index 5c080ac64c9a..abc5c348331d 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -50,11 +50,13 @@ * *

The commit is gated on the plan's done-timestamp watermark. * - *

Watermarks are forwarded only after the cycle commits, never mid-cycle. The {@link - * LockRemover} releases the maintenance lock once a watermark past the trigger's start epoch - * reaches it. The planner emits phase watermarks in the middle of a cycle; forwarding those would - * release the lock before this commit, letting the TriggerManager start a concurrent cycle that - * re-processes the same uncommitted staging snapshot. + *

Watermarks are forwarded while no cycle is active, so a watermark from another task's + * trigger sharing the maintenance lock is not stalled. Once the planner marks cycle start, + * watermarks are held back until the cycle commits, never mid-cycle. The {@link LockRemover} + * releases the maintenance lock once a watermark past the trigger's start epoch reaches it. The + * planner emits phase watermarks before the plan arrives; forwarding those would release the lock + * before this commit, letting the TriggerManager start a concurrent cycle that re-processes the + * same uncommitted staging snapshot. * *

Emits a {@link Trigger} after each cycle (commit, no-op, or error) so the downstream {@link * TaskResultAggregator} can track task completion. This is the sole source of Trigger records for @@ -93,6 +95,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator private transient Table table; private transient List bufferedResults; private transient EqualityConvertPlan planResult; + private transient boolean cycleInFlight; private transient Counter errorCounter; private transient Counter addedDataFileNumCounter; @@ -143,24 +146,38 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { + if (record.getValue().cycleStart()) { + // The planner started a conversion cycle. Hold back watermarks until the plan arrives; the + // planner emits phase watermarks before the plan, and forwarding those would release the + // maintenance lock before the commit. + cycleInFlight = true; + return; + } + planResult = record.getValue(); + cycleInFlight = true; } @Override public void processWatermark(Watermark mark) throws Exception { - if (planResult == null) { + if (!cycleInFlight) { // No conversion cycle is active: forward the watermark. It may originate from another - // task's trigger sharing the maintenance lock, and holding it back would stall lock - // release for that task. + // task's trigger sharing the maintenance lock, and holding it back would stall lock release + // for that task. super.processWatermark(mark); return; } + if (planResult == null) { + // The cycle started but the plan has not arrived yet. The planner emits phase watermarks + // before the plan; forwarding them would release the maintenance lock before the commit, + // letting the next trigger run a concurrent cycle on the same staging snapshot. + return; + } + if (mark.getTimestamp() < planResult.doneTimestamp()) { - // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock - // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the - // lock early and could let the next trigger run a concurrent cycle on the same staging - // snapshot. + // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance + // lock for the whole cycle. return; } @@ -178,6 +195,7 @@ public void processWatermark(Watermark mark) throws Exception { bufferedResults.clear(); planResult = null; + cycleInFlight = false; super.processWatermark(mark); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java index 486b933ee6d7..f506fba8cf2f 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java @@ -133,6 +133,10 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { + if (record.getValue().cycleStart()) { + return; + } + planResult = record.getValue(); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index 191bf8322039..ec4a150faef4 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -34,6 +34,12 @@ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to construct one and {@link * #noOp()} to check. * + *

A cycle-start marker is encoded by {@link #stagingSnapshotId()} == {@link + * #CYCLE_START_STAGING_SNAPSHOT_ID}. The planner emits one on the metadata stream before the + * first phase watermark so the committer holds back watermarks until the plan arrives. Use + * {@link #cycleStart(long)} to construct one and {@link #cycleStart()} to check. It is never a + * plan: consumers must not treat it as one. + * * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with * newly-created DVs @@ -60,12 +66,18 @@ public record EqualityConvertPlan( implements Serializable { public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L; + public static final long CYCLE_START_STAGING_SNAPSHOT_ID = -2L; /** Returns {@code true} if this cycle has nothing to commit. */ public boolean noOp() { return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID; } + /** Returns {@code true} if this is a cycle-start marker rather than a plan. */ + public boolean cycleStart() { + return stagingSnapshotId == CYCLE_START_STAGING_SNAPSHOT_ID; + } + /** No-op cycle: empty file lists, sentinel staging snapshot id. */ public static EqualityConvertPlan noOp( Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) { @@ -78,4 +90,19 @@ public static EqualityConvertPlan noOp( triggerTimestamp, doneTimestamp); } + + /** + * Cycle-start marker. The done timestamp is maximal so no watermark can act on it if a consumer + * fails to filter it out. + */ + public static EqualityConvertPlan cycleStart(long triggerTimestamp) { + return new EqualityConvertPlan( + ImmutableList.of(), + ImmutableList.of(), + ImmutableList.of(), + CYCLE_START_STAGING_SNAPSHOT_ID, + null, + triggerTimestamp, + Long.MAX_VALUE); + } } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java index 82791bae4bc1..3af1b705fd31 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java @@ -82,7 +82,9 @@ * {@link #advancePhase()}. * *

An {@link EqualityConvertPlan} with the current cycle's metadata is emitted via the {@link - * #METADATA_STREAM} side output after the read commands. + * #METADATA_STREAM} side output after the read commands. A {@link EqualityConvertPlan#cycleStart} + * marker is emitted first, before any phase watermark, so the committer can hold back watermarks + * until the plan arrives. * *

Assumes a single equality-field set supplied via the builder; staging eq-deletes with a * different {@code equalityFieldIds} fail fast in {@link #retrieveStagingFiles}. Concurrent writes @@ -264,6 +266,12 @@ public void processElement(StreamRecord element) throws Exception { long triggerTs = element.getTimestamp(); nextPhaseTs = Math.max(triggerTs, nextPhaseTs + 1); + // Mark cycle start before the first phase watermark. The committer holds back watermarks from + // here until the plan arrives; without the marker it cannot tell a phase watermark emitted + // before the plan from an idle watermark that must be forwarded, and forwarding the former + // would release the maintenance lock before the commit. + output.collect(METADATA_STREAM, new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); + Long currentMainSnapshotId = lastMainSnapshotId; try { table.refresh(); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 1bca49a39fc8..35aa9908bce7 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -133,6 +133,49 @@ void holdsBackWatermarkUntilCommit() throws Exception { } } + @Test + void holdsBackPhaseWatermarkBeforePlanArrives() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + insert(table, 1, "a"); + + DataFile stagingDataFile = getFirstDataFile(table); + long snapshotIdBefore = table.currentSnapshot().snapshotId(); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + long triggerTs = System.currentTimeMillis(); + long doneTs = triggerTs + 10; + + // The planner marks cycle start, then emits phase watermarks before the plan. The pre-plan + // phase watermark must not escape: forwarding it would let the LockRemover release the + // maintenance lock before this cycle commits. + harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); + harness.processBothWatermarks(new Watermark(triggerTs + 1)); + assertThat(watermarks(harness)).isEmpty(); + assertThat(harness.extractOutputValues()).isEmpty(); + + // The plan arrives; the done-timestamp watermark then commits the cycle and forwards. + EqualityConvertPlan planResult = + new EqualityConvertPlan( + Lists.newArrayList(stagingDataFile), + Lists.newArrayList(), + Lists.newArrayList(), + 42L, + snapshotIdBefore, + triggerTs, + doneTs); + harness.processElement2(new StreamRecord<>(planResult)); + harness.processBothWatermarks(new Watermark(doneTs)); + + assertThat(harness.extractOutputValues()).hasSize(1); + assertThat(watermarks(harness)).containsExactly(new Watermark(doneTs)); + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()).isNotEqualTo(snapshotIdBefore); + } + } + @Test void forwardsWatermarkWithoutActiveCycle() throws Exception { Table table = createTable(3, FileFormat.PARQUET); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java index 56c6a2f98b33..3f49ad3fa9b6 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java @@ -137,6 +137,36 @@ void noOutputWithoutPlanResult() throws Exception { } } + @Test + void ignoresCycleStartMarker() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + String dataFilePath = getFirstDataFilePath(table); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + long time = System.currentTimeMillis(); + harness.processElement1( + new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time)); + + // The cycle-start marker is not a plan: it must not trigger a write. + harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(time), time)); + harness.processBothWatermarks(new Watermark(time)); + assertThat(harness.extractOutputValues()).isEmpty(); + + // The real plan that follows still drives the write. + harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time)); + harness.processBothWatermarks(new Watermark(time)); + + List output = harness.extractOutputValues(); + assertThat(output).hasSize(1); + assertThat(output.get(0).hasError()).isFalse(); + } + } + @Test void writesDVFilesForMultipleDataFiles() throws Exception { Table table = createTableWithDelete(3); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java index ed243695bced..08e536388adf 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java @@ -84,7 +84,7 @@ void bootstrapsIndexFromBuilderEqualityFieldIds() throws Exception { assertThat(countDataFileTasks(commands)).isEqualTo(2); assertThat(countEqDeleteTasks(commands)).isZero(); - assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2); } } @@ -202,8 +202,9 @@ void emitsReadCommandsForEqualityDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - assertThat(metadata.get(0).getValue().dataFiles()).isEmpty(); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + assertThat(metadata.get(1).getValue().dataFiles()).isEmpty(); } } @@ -280,8 +281,9 @@ void includesNewDataFilesFromStaging() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - assertThat(metadata.get(0).getValue().dataFiles()).hasSize(1); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + assertThat(metadata.get(1).getValue().dataFiles()).hasSize(1); } } @@ -371,7 +373,7 @@ void noOutputWhenStagingBranchEmpty() throws Exception { // emitted because there's nothing on staging to convert. assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(1); assertThat(countEqDeleteTasks(harness.extractOutputValues())).isZero(); - assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2); assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1); assertThat(planner(harness).reindexCount()).isZero(); } @@ -405,8 +407,9 @@ void propagatesStagingOnlyPositionalDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - EqualityConvertPlan result = metadata.get(0).getValue(); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + EqualityConvertPlan result = metadata.get(1).getValue(); assertThat(result.dataFiles()).isEmpty(); assertThat(result.stagingDVFiles()).hasSize(1); assertThat(result.stagingDVFiles().get(0).location()).isEqualTo(stagingDV.location()); From 049260595d76866ec821468b1e46a9d9fc474191 Mon Sep 17 00:00:00 2001 From: Rakesh Pai <41351936+developer-rpai@users.noreply.github.com> Date: Sun, 11 Oct 2026 22:07:53 -0700 Subject: [PATCH 3/8] Fix spotless Javadoc formatting violations Reflow Javadoc paragraphs in EqualityConvertPlan and EqualityConvertCommitter to match palantir-java-format. --- .../maintenance/operator/EqualityConvertCommitter.java | 10 +++------- .../maintenance/operator/EqualityConvertPlan.java | 6 ++---- 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index abc5c348331d..8186269e85ce 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -50,13 +50,9 @@ * *

The commit is gated on the plan's done-timestamp watermark. * - *

Watermarks are forwarded while no cycle is active, so a watermark from another task's - * trigger sharing the maintenance lock is not stalled. Once the planner marks cycle start, - * watermarks are held back until the cycle commits, never mid-cycle. The {@link LockRemover} - * releases the maintenance lock once a watermark past the trigger's start epoch reaches it. The - * planner emits phase watermarks before the plan arrives; forwarding those would release the lock - * before this commit, letting the TriggerManager start a concurrent cycle that re-processes the - * same uncommitted staging snapshot. + *

Watermarks are forwarded while no cycle is active. After the planner marks cycle start, + * watermarks are held back until the cycle commits; forwarding a pre-plan phase watermark would + * let the {@link LockRemover} release the maintenance lock before this commit. * *

Emits a {@link Trigger} after each cycle (commit, no-op, or error) so the downstream {@link * TaskResultAggregator} can track task completion. This is the sole source of Trigger records for diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index ec4a150faef4..33561ce54300 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -35,10 +35,8 @@ * #noOp()} to check. * *

A cycle-start marker is encoded by {@link #stagingSnapshotId()} == {@link - * #CYCLE_START_STAGING_SNAPSHOT_ID}. The planner emits one on the metadata stream before the - * first phase watermark so the committer holds back watermarks until the plan arrives. Use - * {@link #cycleStart(long)} to construct one and {@link #cycleStart()} to check. It is never a - * plan: consumers must not treat it as one. + * #CYCLE_START_STAGING_SNAPSHOT_ID}. Use {@link #cycleStart(long)} to construct one and {@link + * #cycleStart()} to check. It is never a plan: consumers must not treat it as one. * * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with From 9dcc38d61cb9a6b2a0aadd2a8f44da003eefbe34 Mon Sep 17 00:00:00 2001 From: Rakesh Pai <41351936+developer-rpai@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:01:52 -0700 Subject: [PATCH 4/8] Fix remaining spotless Javadoc wrapping in EqualityConvertCommitter --- .../flink/maintenance/operator/EqualityConvertCommitter.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index 8186269e85ce..96db60923cfc 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -51,8 +51,8 @@ *

The commit is gated on the plan's done-timestamp watermark. * *

Watermarks are forwarded while no cycle is active. After the planner marks cycle start, - * watermarks are held back until the cycle commits; forwarding a pre-plan phase watermark would - * let the {@link LockRemover} release the maintenance lock before this commit. + * watermarks are held back until the cycle commits; forwarding a pre-plan phase watermark would let + * the {@link LockRemover} release the maintenance lock before this commit. * *

Emits a {@link Trigger} after each cycle (commit, no-op, or error) so the downstream {@link * TaskResultAggregator} can track task completion. This is the sole source of Trigger records for From bb7cc3f8715aceeeec3911ba8e5357dcde9658f1 Mon Sep 17 00:00:00 2001 From: Rakesh Pai <41351936+developer-rpai@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:44:38 -0700 Subject: [PATCH 5/8] Flink: Advance watermark past marker phase in DVWriter marker test --- .../maintenance/operator/TestEqualityConvertDVWriter.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java index 3f49ad3fa9b6..93614fe8ab11 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java @@ -157,9 +157,10 @@ void ignoresCycleStartMarker() throws Exception { harness.processBothWatermarks(new Watermark(time)); assertThat(harness.extractOutputValues()).isEmpty(); - // The real plan that follows still drives the write. + // The real plan that follows still drives the write. The watermark must advance past the + // previous one for the operator to process it. harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time)); - harness.processBothWatermarks(new Watermark(time)); + harness.processBothWatermarks(new Watermark(time + 1)); List output = harness.extractOutputValues(); assertThat(output).hasSize(1); From cef37c032693015f90da12e6f58c20d8c0cd5673 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 01:04:42 +0000 Subject: [PATCH 6/8] Flink: Hold equality-convert watermarks with the cycle-start marker Drop the cycleInFlight flag. The committer stores the planner's cycle-start marker as the plan result so its maximal done timestamp holds phase watermarks until the real plan arrives, and forwards watermarks when no cycle is active. An end-of-input watermark while the marker is current is forwarded without a commit. Co-authored-by: Rakesh Pai Generated-by: Cursor --- .../operator/EqualityConvertCommitter.java | 47 +++--- .../operator/EqualityConvertPlanner.java | 7 +- .../TestConvertEqualityDeletesSharedLock.java | 151 ++++++++++++++++++ .../TestEqualityConvertCommitter.java | 21 +++ 4 files changed, 198 insertions(+), 28 deletions(-) create mode 100644 flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index 96db60923cfc..c7a8e9c58404 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -50,9 +50,11 @@ * *

The commit is gated on the plan's done-timestamp watermark. * - *

Watermarks are forwarded while no cycle is active. After the planner marks cycle start, - * watermarks are held back until the cycle commits; forwarding a pre-plan phase watermark would let - * the {@link LockRemover} release the maintenance lock before this commit. + *

Watermarks are forwarded only after the cycle commits, never mid-cycle. The {@link + * LockRemover} releases the maintenance lock once a watermark past the trigger's start epoch + * reaches it. The planner emits phase watermarks in the middle of a cycle; forwarding those would + * release the lock before this commit, letting the TriggerManager start a concurrent cycle that + * re-processes the same uncommitted staging snapshot. * *

Emits a {@link Trigger} after each cycle (commit, no-op, or error) so the downstream {@link * TaskResultAggregator} can track task completion. This is the sole source of Trigger records for @@ -91,7 +93,6 @@ public class EqualityConvertCommitter extends AbstractStreamOperator private transient Table table; private transient List bufferedResults; private transient EqualityConvertPlan planResult; - private transient boolean cycleInFlight; private transient Counter errorCounter; private transient Counter addedDataFileNumCounter; @@ -142,38 +143,35 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { - if (record.getValue().cycleStart()) { - // The planner started a conversion cycle. Hold back watermarks until the plan arrives; the - // planner emits phase watermarks before the plan, and forwarding those would release the - // maintenance lock before the commit. - cycleInFlight = true; - return; - } - + // Includes the cycle-start marker. Its maximal done timestamp holds watermarks until the real + // plan replaces it. planResult = record.getValue(); - cycleInFlight = true; } @Override public void processWatermark(Watermark mark) throws Exception { - if (!cycleInFlight) { - // No conversion cycle is active: forward the watermark. It may originate from another - // task's trigger sharing the maintenance lock, and holding it back would stall lock release - // for that task. + if (planResult == null) { + // No cycle in flight: the watermark comes from another task's trigger or a recovery trigger + // sharing the maintenance lock. Holding it back would keep that lock forever. super.processWatermark(mark); return; } - if (planResult == null) { - // The cycle started but the plan has not arrived yet. The planner emits phase watermarks - // before the plan; forwarding them would release the maintenance lock before the commit, - // letting the next trigger run a concurrent cycle on the same staging snapshot. + if (mark.getTimestamp() < planResult.doneTimestamp()) { + // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock + // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the + // lock early and could let the next trigger run a concurrent cycle on the same staging + // snapshot. return; } - if (mark.getTimestamp() < planResult.doneTimestamp()) { - // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance - // lock for the whole cycle. + if (planResult.cycleStart()) { + // The marker's done timestamp is maximal, so only an end-of-input watermark reaches here + // before the plan replaces the marker. Forward that watermark and drop the marker; there is + // nothing to commit. + bufferedResults.clear(); + planResult = null; + super.processWatermark(mark); return; } @@ -191,7 +189,6 @@ public void processWatermark(Watermark mark) throws Exception { bufferedResults.clear(); planResult = null; - cycleInFlight = false; super.processWatermark(mark); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java index 3af1b705fd31..96e5861f1c6f 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java @@ -82,9 +82,10 @@ * {@link #advancePhase()}. * *

An {@link EqualityConvertPlan} with the current cycle's metadata is emitted via the {@link - * #METADATA_STREAM} side output after the read commands. A {@link EqualityConvertPlan#cycleStart} - * marker is emitted first, before any phase watermark, so the committer can hold back watermarks - * until the plan arrives. + * #METADATA_STREAM} side output after the read commands. The planner emits phase watermarks before + * that plan. A {@link EqualityConvertPlan#cycleStart} marker is emitted first, before any phase + * watermark. The marker tells the committer a cycle has started, and the committer holds watermarks + * until the commit. * *

Assumes a single equality-field set supplied via the builder; staging eq-deletes with a * different {@code equalityFieldIds} fail fast in {@link #retrieveStagingFiles}. Concurrent writes diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java new file mode 100644 index 000000000000..c43cf3548361 --- /dev/null +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java @@ -0,0 +1,151 @@ +/* + * 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.iceberg.flink.maintenance.api; + +import static org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME; +import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.Serializable; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.Table; +import org.apache.iceberg.flink.maintenance.operator.ManualSource; +import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests; +import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase; +import org.apache.iceberg.flink.maintenance.operator.TableChange; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +/** + * ConvertEqualityDeletes shares the Flink-maintained maintenance lock with another task. The lock + * is released only when the unioned watermark reaches the lock remover, so a watermark from the + * other task has to pass through an idle converter. + */ +class TestConvertEqualityDeletesSharedLock extends OperatorTestBase { + private static final String PROBE_TASK_NAME = "SharedLockProbe"; + + @Test + void releasesSharedLockAfterEachCycle() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + String tableName = table.name(); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + ManualSource changes = + new ManualSource<>(env, TypeInformation.of(TableChange.class)); + + TableMaintenance.forChangeStream(changes.dataStream(), tableLoader()) + .uidSuffix("ConvertEqualityDeletesSharedLock") + .rateLimit(Duration.ofMillis(50)) + .lockCheckDelay(Duration.ofMillis(50)) + .add(new CompletingTaskBuilder().scheduleOnCommitCount(1)) + .add( + ConvertEqualityDeletes.builder() + .scheduleOnCommitCount(1) + .stagingBranch(SnapshotRef.MAIN_BRANCH) + .targetBranch(SnapshotRef.MAIN_BRANCH) + .equalityFieldColumns(ImmutableList.of("id")) + .parallelism(1)) + .append(); + + List probeSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + PROBE_TASK_NAME + " [0]", + "0", + SUCCEEDED_TASK_COUNTER); + List converterSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + "ConvertEqualityDeletes [1]", + "1", + SUCCEEDED_TASK_COUNTER); + + JobClient jobClient = null; + try { + jobClient = env.executeAsync(); + + // The probe runs first, while the converter has no plan. Its watermark must leave the + // converter or the shared lock stays held and the converter never starts. + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 1L); + awaitSucceeded(converterSucceeded, 1L); + + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 2L); + awaitSucceeded(converterSucceeded, 2L); + } finally { + closeJobClient(jobClient); + } + } + + private static void awaitSucceeded(List counterKey, long expected) { + Awaitility.await() + .atMost(Duration.ofMinutes(3)) + .pollInterval(Duration.ofMillis(200)) + .untilAsserted( + () -> + assertThat(MetricsReporterFactoryForTests.counter(counterKey)).isEqualTo(expected)); + } + + private static class CompletingTaskBuilder extends MaintenanceTaskBuilder { + + @Override + String maintenanceTaskName() { + return PROBE_TASK_NAME; + } + + @Override + DataStream append(DataStream trigger) { + return setSlotSharingGroup( + trigger + .map(new CompletingTask()) + .name(maintenanceTaskName()) + .uid(uidSuffix() + "-shared-lock-probe") + .forceNonParallel()); + } + } + + private static class CompletingTask + implements MapFunction, ResultTypeQueryable, Serializable { + + @Override + public TaskResult map(Trigger trigger) { + return new TaskResult(trigger.taskId(), trigger.timestamp(), true, Collections.emptyList()); + } + + @Override + public TypeInformation getProducedType() { + return TypeInformation.of(TaskResult.class); + } + } +} diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 35aa9908bce7..4fef09caf2c8 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -176,6 +176,27 @@ void holdsBackPhaseWatermarkBeforePlanArrives() throws Exception { } } + @Test + void forwardsMaxWatermarkForCycleStartMarker() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + insert(table, 1, "a"); + long snapshotIdBefore = table.currentSnapshot().snapshotId(); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + long triggerTs = System.currentTimeMillis(); + harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); + harness.processBothWatermarks(Watermark.MAX_WATERMARK); + + assertThat(harness.extractOutputValues()).isEmpty(); + assertThat(watermarks(harness)).containsExactly(Watermark.MAX_WATERMARK); + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()).isEqualTo(snapshotIdBefore); + } + } + @Test void forwardsWatermarkWithoutActiveCycle() throws Exception { Table table = createTable(3, FileFormat.PARQUET); From 67432fed68dee8a0d0eb86e60e4b4e21f51e369c Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Tue, 20 Oct 2026 09:25:58 +0200 Subject: [PATCH 7/8] fixup! Flink: Encode the equality convert cycle-start marker as a no-op plan The marker's done timestamp is never reached, which already holds back watermarks until the cycle's actual plan replaces it. --- .../operator/EqualityConvertCommitter.java | 10 -------- .../operator/EqualityConvertDVWriter.java | 4 ---- .../operator/EqualityConvertPlan.java | 24 ++++--------------- .../TestEqualityConvertCommitter.java | 21 ---------------- .../operator/TestEqualityConvertPlanner.java | 6 ++--- 5 files changed, 8 insertions(+), 57 deletions(-) diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index c7a8e9c58404..34d996c3d9c3 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -165,16 +165,6 @@ public void processWatermark(Watermark mark) throws Exception { return; } - if (planResult.cycleStart()) { - // The marker's done timestamp is maximal, so only an end-of-input watermark reaches here - // before the plan replaces the marker. Forward that watermark and drop the marker; there is - // nothing to commit. - bufferedResults.clear(); - planResult = null; - super.processWatermark(mark); - return; - } - try { commitIfNeeded(); } catch (Exception e) { diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java index f506fba8cf2f..486b933ee6d7 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java @@ -133,10 +133,6 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { - if (record.getValue().cycleStart()) { - return; - } - planResult = record.getValue(); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index 33561ce54300..4aa326d19a19 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -34,9 +34,8 @@ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to construct one and {@link * #noOp()} to check. * - *

A cycle-start marker is encoded by {@link #stagingSnapshotId()} == {@link - * #CYCLE_START_STAGING_SNAPSHOT_ID}. Use {@link #cycleStart(long)} to construct one and {@link - * #cycleStart()} to check. It is never a plan: consumers must not treat it as one. + *

A cycle-start marker is a no-op plan whose done timestamp is never reached. Use {@link + * #cycleStart(long)} to construct one. * * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with @@ -64,18 +63,12 @@ public record EqualityConvertPlan( implements Serializable { public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L; - public static final long CYCLE_START_STAGING_SNAPSHOT_ID = -2L; /** Returns {@code true} if this cycle has nothing to commit. */ public boolean noOp() { return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID; } - /** Returns {@code true} if this is a cycle-start marker rather than a plan. */ - public boolean cycleStart() { - return stagingSnapshotId == CYCLE_START_STAGING_SNAPSHOT_ID; - } - /** No-op cycle: empty file lists, sentinel staging snapshot id. */ public static EqualityConvertPlan noOp( Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) { @@ -90,17 +83,10 @@ public static EqualityConvertPlan noOp( } /** - * Cycle-start marker. The done timestamp is maximal so no watermark can act on it if a consumer - * fails to filter it out. + * Cycle-start marker: a no-op plan whose done timestamp is never reached, so consumers hold it + * until the cycle's actual plan replaces it. */ public static EqualityConvertPlan cycleStart(long triggerTimestamp) { - return new EqualityConvertPlan( - ImmutableList.of(), - ImmutableList.of(), - ImmutableList.of(), - CYCLE_START_STAGING_SNAPSHOT_ID, - null, - triggerTimestamp, - Long.MAX_VALUE); + return noOp(null, triggerTimestamp, Long.MAX_VALUE); } } diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 4fef09caf2c8..35aa9908bce7 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -176,27 +176,6 @@ void holdsBackPhaseWatermarkBeforePlanArrives() throws Exception { } } - @Test - void forwardsMaxWatermarkForCycleStartMarker() throws Exception { - Table table = createTable(3, FileFormat.PARQUET); - insert(table, 1, "a"); - long snapshotIdBefore = table.currentSnapshot().snapshotId(); - - try (TwoInputStreamOperatorTestHarness harness = - createHarness()) { - harness.open(); - - long triggerTs = System.currentTimeMillis(); - harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); - harness.processBothWatermarks(Watermark.MAX_WATERMARK); - - assertThat(harness.extractOutputValues()).isEmpty(); - assertThat(watermarks(harness)).containsExactly(Watermark.MAX_WATERMARK); - table.refresh(); - assertThat(table.currentSnapshot().snapshotId()).isEqualTo(snapshotIdBefore); - } - } - @Test void forwardsWatermarkWithoutActiveCycle() throws Exception { Table table = createTable(3, FileFormat.PARQUET); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java index 08e536388adf..1482ce872661 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java @@ -203,7 +203,7 @@ void emitsReadCommandsForEqualityDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); assertThat(metadata).hasSize(2); - assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); assertThat(metadata.get(1).getValue().dataFiles()).isEmpty(); } } @@ -282,7 +282,7 @@ void includesNewDataFilesFromStaging() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); assertThat(metadata).hasSize(2); - assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); assertThat(metadata.get(1).getValue().dataFiles()).hasSize(1); } } @@ -408,7 +408,7 @@ void propagatesStagingOnlyPositionalDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); assertThat(metadata).hasSize(2); - assertThat(metadata.get(0).getValue().cycleStart()).isTrue(); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); EqualityConvertPlan result = metadata.get(1).getValue(); assertThat(result.dataFiles()).isEmpty(); assertThat(result.stagingDVFiles()).hasSize(1); From 0bd929a02be35113000932f6af8f288243a45ab8 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Tue, 20 Oct 2026 11:28:10 +0200 Subject: [PATCH 8/8] fixup! Flink: Move the shared-lock test into TestConvertEqualityDeletesE2E --- .../api/TestConvertEqualityDeletesE2E.java | 116 ++++++++++++++ .../TestConvertEqualityDeletesSharedLock.java | 151 ------------------ 2 files changed, 116 insertions(+), 151 deletions(-) delete mode 100644 flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java index f3bdaa8fe1c6..e39d05b66f00 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java @@ -19,13 +19,22 @@ package org.apache.iceberg.flink.maintenance.api; import static org.apache.iceberg.flink.SimpleDataUtil.createRecord; +import static org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME; +import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER; import static org.assertj.core.api.Assertions.assertThat; import java.io.File; import java.io.IOException; +import java.io.Serializable; import java.nio.file.Path; import java.time.Duration; +import java.util.Collections; +import java.util.List; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; import org.apache.flink.core.execution.JobClient; +import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; @@ -42,12 +51,16 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericAppenderHelper; import org.apache.iceberg.flink.SimpleDataUtil; +import org.apache.iceberg.flink.maintenance.operator.ManualSource; +import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests; import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase; +import org.apache.iceberg.flink.maintenance.operator.TableChange; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.util.ContentFileUtil; import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -60,6 +73,7 @@ */ class TestConvertEqualityDeletesE2E extends OperatorTestBase { private static final String STAGING_BRANCH = "staging"; + private static final String PROBE_TASK_NAME = "SharedLockProbe"; @TempDir private Path tempDir; private StreamExecutionEnvironment env; @@ -136,6 +150,76 @@ void testConvertEqualityDeletesE2E(String stagingBranch) throws Exception { } } + /** + * ConvertEqualityDeletes shares the Flink-maintained maintenance lock with another task. The lock + * is released only when the unioned watermark reaches the lock remover, so a watermark from the + * other task has to pass through an idle converter. + */ + @Test + void releasesSharedLockAfterEachCycle() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + String tableName = table.name(); + + env.setParallelism(1); + ManualSource changes = + new ManualSource<>(env, TypeInformation.of(TableChange.class)); + + TableMaintenance.forChangeStream(changes.dataStream(), tableLoader()) + .uidSuffix("ConvertEqualityDeletesSharedLock") + .rateLimit(Duration.ofMillis(50)) + .lockCheckDelay(Duration.ofMillis(50)) + .add(new CompletingTaskBuilder().scheduleOnCommitCount(1)) + .add( + ConvertEqualityDeletes.builder() + .scheduleOnCommitCount(1) + .stagingBranch(SnapshotRef.MAIN_BRANCH) + .targetBranch(SnapshotRef.MAIN_BRANCH) + .equalityFieldColumns(ImmutableList.of("id")) + .parallelism(1)) + .append(); + + List probeSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + PROBE_TASK_NAME + " [0]", + "0", + SUCCEEDED_TASK_COUNTER); + List converterSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + "ConvertEqualityDeletes [1]", + "1", + SUCCEEDED_TASK_COUNTER); + + JobClient jobClient = null; + try { + jobClient = env.executeAsync(); + + // The probe runs first, while the converter has no plan. Its watermark must leave the + // converter or the shared lock stays held and the converter never starts. + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 1L); + awaitSucceeded(converterSucceeded, 1L); + + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 2L); + awaitSucceeded(converterSucceeded, 2L); + } finally { + closeJobClient(jobClient); + } + } + + private static void awaitSucceeded(List counterKey, long expected) { + Awaitility.await() + .atMost(Duration.ofMinutes(3)) + .pollInterval(Duration.ofMillis(200)) + .untilAsserted( + () -> + assertThat(MetricsReporterFactoryForTests.counter(counterKey)).isEqualTo(expected)); + } + private DataFile writeDataFile(Table table, Integer id, String data) throws IOException { return new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir) .writeFile(Lists.newArrayList(SimpleDataUtil.createRecord(id, data))); @@ -193,4 +277,36 @@ private static long eqDeleteCountOnMain(Table table) throws IOException { return count; } + + private static class CompletingTaskBuilder extends MaintenanceTaskBuilder { + + @Override + String maintenanceTaskName() { + return PROBE_TASK_NAME; + } + + @Override + DataStream append(DataStream trigger) { + return setSlotSharingGroup( + trigger + .map(new CompletingTask()) + .name(maintenanceTaskName()) + .uid(uidSuffix() + "-shared-lock-probe") + .forceNonParallel()); + } + } + + private static class CompletingTask + implements MapFunction, ResultTypeQueryable, Serializable { + + @Override + public TaskResult map(Trigger trigger) { + return new TaskResult(trigger.taskId(), trigger.timestamp(), true, Collections.emptyList()); + } + + @Override + public TypeInformation getProducedType() { + return TypeInformation.of(TaskResult.class); + } + } } diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java deleted file mode 100644 index c43cf3548361..000000000000 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesSharedLock.java +++ /dev/null @@ -1,151 +0,0 @@ -/* - * 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.iceberg.flink.maintenance.api; - -import static org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME; -import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER; -import static org.assertj.core.api.Assertions.assertThat; - -import java.io.Serializable; -import java.time.Duration; -import java.util.Collections; -import java.util.List; -import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.common.typeinfo.TypeInformation; -import org.apache.flink.api.java.typeutils.ResultTypeQueryable; -import org.apache.flink.core.execution.JobClient; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.SnapshotRef; -import org.apache.iceberg.Table; -import org.apache.iceberg.flink.maintenance.operator.ManualSource; -import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests; -import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase; -import org.apache.iceberg.flink.maintenance.operator.TableChange; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; -import org.awaitility.Awaitility; -import org.junit.jupiter.api.Test; - -/** - * ConvertEqualityDeletes shares the Flink-maintained maintenance lock with another task. The lock - * is released only when the unioned watermark reaches the lock remover, so a watermark from the - * other task has to pass through an idle converter. - */ -class TestConvertEqualityDeletesSharedLock extends OperatorTestBase { - private static final String PROBE_TASK_NAME = "SharedLockProbe"; - - @Test - void releasesSharedLockAfterEachCycle() throws Exception { - Table table = createTable(3, FileFormat.PARQUET); - String tableName = table.name(); - - StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - env.setParallelism(1); - ManualSource changes = - new ManualSource<>(env, TypeInformation.of(TableChange.class)); - - TableMaintenance.forChangeStream(changes.dataStream(), tableLoader()) - .uidSuffix("ConvertEqualityDeletesSharedLock") - .rateLimit(Duration.ofMillis(50)) - .lockCheckDelay(Duration.ofMillis(50)) - .add(new CompletingTaskBuilder().scheduleOnCommitCount(1)) - .add( - ConvertEqualityDeletes.builder() - .scheduleOnCommitCount(1) - .stagingBranch(SnapshotRef.MAIN_BRANCH) - .targetBranch(SnapshotRef.MAIN_BRANCH) - .equalityFieldColumns(ImmutableList.of("id")) - .parallelism(1)) - .append(); - - List probeSucceeded = - ImmutableList.of( - LOCK_REMOVER_OPERATOR_NAME, - tableName, - PROBE_TASK_NAME + " [0]", - "0", - SUCCEEDED_TASK_COUNTER); - List converterSucceeded = - ImmutableList.of( - LOCK_REMOVER_OPERATOR_NAME, - tableName, - "ConvertEqualityDeletes [1]", - "1", - SUCCEEDED_TASK_COUNTER); - - JobClient jobClient = null; - try { - jobClient = env.executeAsync(); - - // The probe runs first, while the converter has no plan. Its watermark must leave the - // converter or the shared lock stays held and the converter never starts. - changes.sendRecord(TableChange.builder().commitCount(1).build()); - awaitSucceeded(probeSucceeded, 1L); - awaitSucceeded(converterSucceeded, 1L); - - changes.sendRecord(TableChange.builder().commitCount(1).build()); - awaitSucceeded(probeSucceeded, 2L); - awaitSucceeded(converterSucceeded, 2L); - } finally { - closeJobClient(jobClient); - } - } - - private static void awaitSucceeded(List counterKey, long expected) { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofMillis(200)) - .untilAsserted( - () -> - assertThat(MetricsReporterFactoryForTests.counter(counterKey)).isEqualTo(expected)); - } - - private static class CompletingTaskBuilder extends MaintenanceTaskBuilder { - - @Override - String maintenanceTaskName() { - return PROBE_TASK_NAME; - } - - @Override - DataStream append(DataStream trigger) { - return setSlotSharingGroup( - trigger - .map(new CompletingTask()) - .name(maintenanceTaskName()) - .uid(uidSuffix() + "-shared-lock-probe") - .forceNonParallel()); - } - } - - private static class CompletingTask - implements MapFunction, ResultTypeQueryable, Serializable { - - @Override - public TaskResult map(Trigger trigger) { - return new TaskResult(trigger.taskId(), trigger.timestamp(), true, Collections.emptyList()); - } - - @Override - public TypeInformation getProducedType() { - return TypeInformation.of(TaskResult.class); - } - } -}