(iceberg) branch main updated: Flink: Hold back equality delete converter watermark until completion (#17038)

[hu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 49b89a8c59 Flink: Hold back equality delete converter watermark until completion (#17038)
49b89a8c59 is described below

commit 49b89a8c59d7d88290c6e925f41b65fa9fc99a88
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Fri Jul 3 08:33:17 2026 +0200

    Flink: Hold back equality delete converter watermark until completion (#17038)
    
    This PR fixes the CI test flakiness seen in:
    
    ```
    TestConvertEqualityDeletesE2E > testConvertEqualityDeletesE2E(String) > [1] staging FAILED
        org.opentest4j.AssertionFailedError:
        expected: 2L
         but was: 1L
            at app//org.apache.iceberg.flink.maintenance.api.TestConvertEqualityDeletesE2E.lambda$testConvertEqualityDeletesE2E$1(TestConvertEqualityDeletesE2E.java:127)
    ```
    e.g.:
    https://github.com/apache/iceberg/actions/runs/28499724055/job/84473933155?pr=16293
    
    Flink's table maintenance framework maintains a lock to prevent concurrent
    execution of maintenance tasks. The component responsible for removing the
    lock (LockRemover) releases the task lock once a watermark reaches it past the
    task's start timestamp.
    
    EqualityConvertPlanner emits phase watermarks in the middle of its execution,
    and the committer forwarded them, so the lock was released before the run
    completed. The maintenance framework then started a next cycle concurrently, and
    both re-processed the same uncommitted staging snapshot. The overlapping commits
    conflicted, and one advanced its commit marker while dropping its deletion
    vector, causing the test flakiness.
    
    The solution is to forward watermarks from the committer only after it finishes
    the conversion cycle, to ensure mutual exclusive execution of the maintenance
    tasks.
---
 .../operator/EqualityConvertCommitter.java         | 44 ++++++++++++---------
 .../operator/TestEqualityConvertCommitter.java     | 45 ++++++++++++++++++++++
 2 files changed, 71 insertions(+), 18 deletions(-)

diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index b3e8fb4d89..95c9ccd5de 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -50,6 +50,12 @@ import org.slf4j.LoggerFactory;
  *
  * <p>The commit is gated on the plan's done-timestamp watermark.
  *
+ * <p>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.
+ *
  * <p>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
  * the Aggregator.
@@ -139,27 +145,29 @@ public class EqualityConvertCommitter extends AbstractStreamOperator<Trigger>
 
   @Override
   public void processWatermark(Watermark mark) throws Exception {
-    if (planResult != null && mark.getTimestamp() >= planResult.doneTimestamp()) {
-      try {
-        commitIfNeeded();
-      } catch (Exception e) {
-        LOG.error(
-            "Failed to commit equality convert result for table {} task {}",
-            tableName,
-            taskName,
-            e);
-        output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
-        errorCounter.inc();
-      }
-
-      // Emit Trigger for the Aggregator (even on error or no-op).
-      output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+    if (planResult == null || 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;
+    }
 
-      bufferedResults.clear();
-      planResult = null;
+    try {
+      commitIfNeeded();
+    } catch (Exception e) {
+      LOG.error(
+          "Failed to commit equality convert result for table {} task {}", tableName, taskName, e);
+      output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+      errorCounter.inc();
     }
 
-    // Always forward watermarks to prevent stalling downstream.
+    // Emit Trigger for the Aggregator (even on error or no-op).
+    output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+
+    bufferedResults.clear();
+    planResult = null;
+
     super.processWatermark(mark);
   }
 
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index f74570c878..54f0b1758d 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.stream.Collectors;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -83,6 +84,43 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
     }
   }
 
+  @Test
+  void holdsBackWatermarkUntilCommit() throws Exception {
+    Table table = createTable(3, FileFormat.PARQUET);
+    insert(table, 1, "a");
+
+    DataFile stagingDataFile = getFirstDataFile(table);
+    long snapshotIdBefore = table.currentSnapshot().snapshotId();
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, Trigger> harness =
+        createHarness()) {
+      harness.open();
+
+      long doneTs = System.currentTimeMillis();
+      EqualityConvertPlan planResult =
+          new EqualityConvertPlan(
+              Lists.newArrayList(stagingDataFile),
+              Lists.newArrayList(),
+              42L,
+              snapshotIdBefore,
+              doneTs - 2,
+              doneTs);
+
+      harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
+
+      // A phase watermark before the plan's done timestamp must not be forwarded: it would let the
+      // LockRemover release the maintenance lock before this cycle commits.
+      harness.processBothWatermarks(new Watermark(doneTs - 1));
+      assertThat(harness.extractOutputValues()).isEmpty();
+      assertThat(watermarks(harness)).isEmpty();
+
+      // The done-timestamp watermark commits the cycle; the watermark is forwarded only now.
+      harness.processBothWatermarks(new Watermark(doneTs));
+      assertThat(harness.extractOutputValues()).hasSize(1);
+      assertThat(watermarks(harness)).containsExactly(new Watermark(doneTs));
+    }
+  }
+
   @Test
   void skipsCommitForEmptyCycle() throws Exception {
     Table table = createTable(3, FileFormat.PARQUET);
@@ -476,6 +514,13 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
             SnapshotRef.MAIN_BRANCH));
   }
 
+  private static List<Watermark> watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
+    return harness.getOutput().stream()
+        .filter(Watermark.class::isInstance)
+        .map(Watermark.class::cast)
+        .collect(Collectors.toList());
+  }
+
   private static List<DeleteFile> deletesForDataFile(Table table, String dataFilePath) {
     List<DeleteFile> deletes = Lists.newArrayList();
     for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {