Flink: Forward watermarks in EqualityConvertCommitter with no active cycle

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java

@@ -143,12 +143,21 @@ public void processElement1(StreamRecord<DVWriteResult> record) {
 
   @Override
   public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+    // Includes the cycle-start marker. Its maximal done timestamp holds watermarks until the real
+    // plan replaces it.
     planResult = record.getValue();
   }
 
   @Override
   public void processWatermark(Watermark mark) throws Exception {
-    if (planResult == null || mark.getTimestamp() < planResult.doneTimestamp()) {
+    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 (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

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java

@@ -34,6 +34,9 @@
  * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to construct one and {@link
  * #noOp()} to check.
  *
+ * <p>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
  *     newly-created DVs
@@ -78,4 +81,12 @@ public static EqualityConvertPlan noOp(
         triggerTimestamp,
         doneTimestamp);
   }
+
+  /**
+   * 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 noOp(null, triggerTimestamp, Long.MAX_VALUE);
+  }
 }

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java

@@ -82,7 +82,10 @@
  * {@link #advancePhase()}.
  *
  * <p>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. 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.
  *
  * <p>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 +267,12 @@ public void processElement(StreamRecord<Trigger> 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();

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<TableChange> 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<String> probeSucceeded =
+        ImmutableList.of(
+            LOCK_REMOVER_OPERATOR_NAME,
+            tableName,
+            PROBE_TASK_NAME + " [0]",
+            "0",
+            SUCCEEDED_TASK_COUNTER);
+    List<String> 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<String> 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<CompletingTaskBuilder> {
+
+    @Override
+    String maintenanceTaskName() {
+      return PROBE_TASK_NAME;
+    }
+
+    @Override
+    DataStream<TaskResult> append(DataStream<Trigger> trigger) {
+      return setSlotSharingGroup(
+          trigger
+              .map(new CompletingTask())
+              .name(maintenanceTaskName())
+              .uid(uidSuffix() + "-shared-lock-probe")
+              .forceNonParallel());
+    }
+  }
+
+  private static class CompletingTask
+      implements MapFunction<Trigger, TaskResult>, ResultTypeQueryable<TaskResult>, Serializable {
+
+    @Override
+    public TaskResult map(Trigger trigger) {
+      return new TaskResult(trigger.taskId(), trigger.timestamp(), true, Collections.emptyList());
+    }
+
+    @Override
+    public TypeInformation<TaskResult> getProducedType() {
+      return TypeInformation.of(TaskResult.class);
+    }
+  }
 }

flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java

@@ -133,6 +133,67 @@ 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<DVWriteResult, EqualityConvertPlan, Trigger> 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);
+    insert(table, 1, "a");
+
+    try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, Trigger> 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);

flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java

@@ -137,6 +137,37 @@ void noOutputWithoutPlanResult() throws Exception {
     }
   }
 
+  @Test
+  void ignoresCycleStartMarker() throws Exception {
+    Table table = createTableWithDelete(3);
+    insert(table, 1, "a");
+
+    String dataFilePath = getFirstDataFilePath(table);
+
+    try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> 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. 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 + 1));
+
+      List<DVWriteResult> output = harness.extractOutputValues();
+      assertThat(output).hasSize(1);
+      assertThat(output.get(0).hasError()).isFalse();
+    }
+  }
+
   @Test
   void writesDVFilesForMultipleDataFiles() throws Exception {
     Table table = createTableWithDelete(3);

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<StreamRecord<EqualityConvertPlan>> 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().doneTimestamp()).isEqualTo(Long.MAX_VALUE);
+      assertThat(metadata.get(1).getValue().dataFiles()).isEmpty();
     }
   }
 
@@ -280,8 +281,9 @@ void includesNewDataFilesFromStaging() throws Exception {
 
       List<StreamRecord<EqualityConvertPlan>> 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().doneTimestamp()).isEqualTo(Long.MAX_VALUE);
+      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<StreamRecord<EqualityConvertPlan>> 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().doneTimestamp()).isEqualTo(Long.MAX_VALUE);
+      EqualityConvertPlan result = metadata.get(1).getValue();
       assertThat(result.dataFiles()).isEmpty();
       assertThat(result.stagingDVFiles()).hasSize(1);
       assertThat(result.stagingDVFiles().get(0).location()).isEqualTo(stagingDV.location());