Repository: apache/iceberg (language: python)

Issue:
Flink: Maintenance - TriggerManager
The responsibility of the Trigger Manager is to start the Maintenance Tasks based on the incoming Table Change messages and prevent overlapping Maintenance Task runs. The event time of the Trigger messages should be monotonically increasing, as this will be used as a watermark to separate Maintenance Task runs.

Implementation for https://github.com/apache/iceberg/issues/10301

This will be used for Maintenance Trigger Manager as described in the [design doc](https://docs.google.com/document/d/16g3vR18mVBy8jbFaLjf2JwAANuYOmIwr15yDDxovdnA). The Trigger Manager paragraph describes the requirements in more detail.


Discussion / hints:
Hi @stevenzwu,

I would like to move forward with this PR.
I see the following open questions:
1. Lock implementation - I still think it is important to provide a working implementation without adding a requirement for external systems. For this the `Tag` based solution is working for every Iceberg deployment and there were no objections or better suggestions for it during the discussion on the mailing list (https://lists.apache.org/thread/vjf8m5wg840o58yz4y3q35k2mfhbm49l). Also tried to ask help from the Flink mailing list for alternative solutions, but I got a single one offline which turned out a dead end (https://lists.apache.org/thread/3ss83w1hp8wpocx13kcqr7qyjnghvnc8). - **I propose to remove the JVM based lock, and move forward with the Tag based lock**
2. TableChange parameters - I am open to discussing the list of parameters which will drive the triggers. It will touch a big part of the previous PR as well, and will make this change even harder to review/read. -  **I propose to do it in the next PR**
3. There were some questions about the overall architecture. This is described in the original proposal doc [here](https://docs.google.com/document/d/16g3vR18mVBy8jbFaLjf2JwAANuYOmIwr15yDDxovdnA/edit#heading=h.5dc7u19et6mc). - **I'm open to discuss the proposal offline too, if that is needed.**

WDYT @stevenzwu? Could we move forward along these lines, or do you have another proposal? Do you have any foundational concerns about the PR? 

Thanks,
Peter
Merged to master.
Thanks for the review @stevenzwu, @singhpk234 and @gyfora!

Fix patch:
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
index f22be33aeae6..085fbfecd270 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
@@ -38,6 +38,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * JDBC table backed implementation of the {@link
+ * org.apache.iceberg.flink.maintenance.operator.TriggerLockFactory}.
+ */
 public class JdbcLockFactory implements TriggerLockFactory {
   private static final Logger LOG = LoggerFactory.getLogger(JdbcLockFactory.class);
 
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
new file mode 100644
index 000000000000..3c3761ef2f4d
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
@@ -0,0 +1,144 @@
+/*
+ * 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.operator;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages locks and collect {@link org.apache.flink.metrics.Metric} for the Maintenance Tasks.
+ *
+ * <p>The assumptions about the locks are the following:
+ *
+ * <ul>
+ *   <li>Every {@link TaskResult} is followed by a {@link Watermark} for normal {@link Trigger}s
+ *   <li>For the {@link Trigger#recovery(long)} {@link Watermark} there is no element to process
+ * </ul>
+ *
+ * When processing the inputs there are 3 possibilities:
+ *
+ * <ul>
+ *   <li>Normal execution - we receive a {@link TaskResult} and then a {@link Watermark} - unlocking
+ *       the lock is handled by the {@link #processElement(StreamRecord)}
+ *   <li>Recovery without ongoing execution (unlocking the recoveryLock) - we receive the {@link
+ *       Trigger#recovery(long)} {@link Watermark} without any {@link TaskResult} - unlocking the
+ *       {@link TriggerLockFactory#createRecoveryLock()} and a possible {@link
+ *       TriggerLockFactory#createLock()} is handled by the {@link #processWatermark(Watermark)}
+ *       (the {@link #lastProcessedTaskStartEpoch} is 0 in this case)
+ *   <li>Recovery with an ongoing execution - we receive a {@link TaskResult} and then a {@link
+ *       Watermark} - unlocking the {@link TriggerLockFactory#createLock()} is handled by the {@link
+ *       #processElement(StreamRecord)}, unlocking the {@link
+ *       TriggerLockFactory#createRecoveryLock()} is handled by the {@link
+ *       #processWatermark(Watermark)} (the {@link #lastProcessedTaskStartEpoch} is the start time
+ *       of the old task)
+ * </ul>
+ */
+@Internal
+public class LockRemover extends AbstractStreamOperator<Void>
+    implements OneInputStreamOperator<TaskResult, Void> {
+  private static final Logger LOG = LoggerFactory.getLogger(LockRemover.class);
+
+  private final TriggerLockFactory lockFactory;
+  private final List<String> maintenanceTaskNames;
+
+  private transient List<Counter> succeededTaskResultCounters;
+  private transient List<Counter> failedTaskResultCounters;
+  private transient List<AtomicLong> taskLastRunDurationMs;
+  private transient TriggerLockFactory.Lock lock;
+  private transient TriggerLockFactory.Lock recoveryLock;
+  private transient long lastProcessedTaskStartEpoch = 0L;
+
+  public LockRemover(TriggerLockFactory lockFactory, List<String> maintenanceTaskNames) {
+    Preconditions.checkNotNull(lockFactory, "Lock factory should no be null");
+    Preconditions.checkArgument(
+        maintenanceTaskNames != null && !maintenanceTaskNames.isEmpty(),
+        "Invalid maintenance task names: null or empty");
+
+    this.lockFactory = lockFactory;
+    this.maintenanceTaskNames = maintenanceTaskNames;
+  }
+
+  @Override
+  public void open() throws Exception {
+    super.open();
+    this.succeededTaskResultCounters =
+        Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
+    this.failedTaskResultCounters = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
+    this.taskLastRunDurationMs = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
+    for (String name : maintenanceTaskNames) {
+      succeededTaskResultCounters.add(
+          getRuntimeContext()
+              .getMetricGroup()
+              .addGroup(TableMaintenanceMetrics.GROUP_KEY, name)
+              .counter(TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER));
+      failedTaskResultCounters.add(
+          getRuntimeContext()
+              .getMetricGroup()
+              .addGroup(TableMaintenanceMetrics.GROUP_KEY, name)
+              .counter(TableMaintenanceMetrics.FAILED_TASK_COUNTER));
+      AtomicLong duration = new AtomicLong(0);
+      taskLastRunDurationMs.add(duration);
+      getRuntimeContext()
+          .getMetricGroup()
+          .addGroup(TableMaintenanceMetrics.GROUP_KEY, name)
+          .gauge(TableMaintenanceMetrics.LAST_RUN_DURATION_MS, duration::get);
+    }
+
+    this.lock = lockFactory.createLock();
+    this.recoveryLock = lockFactory.createRecoveryLock();
+  }
+
+  @Override
+  public void processElement(StreamRecord<TaskResult> streamRecord) {
+    TaskResult taskResult = streamRecord.getValue();
+    LOG.info(
+        "Processing result {} for task {}",
+        taskResult,
+        maintenanceTaskNames.get(taskResult.taskIndex()));
+    long duration = System.currentTimeMillis() - taskResult.startEpoch();
+    lock.unlock();
+    this.lastProcessedTaskStartEpoch = taskResult.startEpoch();
+
+    // Update the metrics
+    taskLastRunDurationMs.get(taskResult.taskIndex()).set(duration);
+    if (taskResult.success()) {
+      succeededTaskResultCounters.get(taskResult.taskIndex()).inc();
+    } else {
+      failedTaskResultCounters.get(taskResult.taskIndex()).inc();
+    }
+  }
+
+  @Override
+  public void processWatermark(Watermark mark) {
+    if (mark.getTimestamp() > lastProcessedTaskStartEpoch) {
+      lock.unlock();
+      recoveryLock.unlock();
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
index ec0fd920c34b..1a04461aed43 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
@@ -28,6 +28,11 @@ public class TableMaintenanceMetrics {
   public static final String TRIGGERED = "triggered";
   public static final String NOTHING_TO_TRIGGER = "nothingToTrigger";
 
+  // LockRemover metrics
+  public static final String SUCCEEDED_TASK_COUNTER = "succeededTasks";
+  public static final String FAILED_TASK_COUNTER = "failedTasks";
+  public static final String LAST_RUN_DURATION_MS = "lastRunDurationMs";
+
   private TableMaintenanceMetrics() {
     // do not instantiate
   }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResult.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResult.java
new file mode 100644
index 000000000000..06f10f1c1d68
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResult.java
@@ -0,0 +1,65 @@
+/*
+ * 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.operator;
+
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+
+/** The result of a single Maintenance Task. */
+@Internal
+public class TaskResult {
+  private final int taskIndex;
+  private final long startEpoch;
+  private final boolean success;
+  private final List<Exception> exceptions;
+
+  public TaskResult(int taskIndex, long startEpoch, boolean success, List<Exception> exceptions) {
+    this.taskIndex = taskIndex;
+    this.startEpoch = startEpoch;
+    this.success = success;
+    this.exceptions = exceptions;
+  }
+
+  public int taskIndex() {
+    return taskIndex;
+  }
+
+  public long startEpoch() {
+    return startEpoch;
+  }
+
+  public boolean success() {
+    return success;
+  }
+
+  public List<Exception> exceptions() {
+    return exceptions;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("taskIndex", taskIndex)
+        .add("startEpoch", startEpoch)
+        .add("success", success)
+        .add("exceptions", exceptions)
+        .toString();
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
index f4c3c1d47cf7..dc95b27af0a6 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
@@ -63,7 +63,7 @@ class TriggerManager extends KeyedProcessFunction<Boolean, TableChange, Trigger>
 
   private final TableLoader tableLoader;
   private final TriggerLockFactory lockFactory;
-  private final List<String> taskNames;
+  private final List<String> maintenanceTaskNames;
   private final List<TriggerEvaluator> evaluators;
   private final long minFireDelayMs;
   private final long lockCheckDelayMs;
@@ -92,25 +92,27 @@ class TriggerManager extends KeyedProcessFunction<Boolean, TableChange, Trigger>
   TriggerManager(
       TableLoader tableLoader,
       TriggerLockFactory lockFactory,
-      List<String> taskNames,
+      List<String> maintenanceTaskNames,
       List<TriggerEvaluator> evaluators,
       long minFireDelayMs,
       long lockCheckDelayMs) {
     Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
     Preconditions.checkNotNull(lockFactory, "Lock factory should no be null");
     Preconditions.checkArgument(
-        taskNames != null && !taskNames.isEmpty(), "Invalid task names: null or empty");
+        maintenanceTaskNames != null && !maintenanceTaskNames.isEmpty(),
+        "Invalid maintenance task names: null or empty");
     Preconditions.checkArgument(
         evaluators != null && !evaluators.isEmpty(), "Invalid evaluators: null or empty");
     Preconditions.checkArgument(
-        taskNames.size() == evaluators.size(), "Provide a name and evaluator for all of the tasks");
+        maintenanceTaskNames.size() == evaluators.size(),
+        "Provide a name and evaluator for all of the maintenance tasks");
     Preconditions.checkArgument(minFireDelayMs > 0, "Minimum fire delay should be at least 1.");
     Preconditions.checkArgument(
         lockCheckDelayMs > 0, "Minimum lock delay rate should be at least 1 ms.");
 
     this.tableLoader = tableLoader;
     this.lockFactory = lockFactory;
-    this.taskNames = taskNames;
+    this.maintenanceTaskNames = maintenanceTaskNames;
     this.evaluators = evaluators;
     this.minFireDelayMs = minFireDelayMs;
     this.lockCheckDelayMs = lockCheckDelayMs;
@@ -137,7 +139,7 @@ public void open(Configuration parameters) throws Exception {
                 TableMaintenanceMetrics.GROUP_KEY, TableMaintenanceMetrics.GROUP_VALUE_DEFAULT)
             .counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
     this.triggerCounters =
-        taskNames.stream()
+        maintenanceTaskNames.stream()
             .map(
                 name ->
                     getRuntimeContext()


Test patch:
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/ManualSource.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/ManualSource.java
index 38bb9c393fa9..679b3ec508a2 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/ManualSource.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/ManualSource.java
@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import org.apache.flink.api.common.eventtime.Watermark;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.source.Boundedness;
@@ -156,23 +157,26 @@ public SimpleVersionedSerializer<DummyCheckpoint> getEnumeratorCheckpointSeriali
 
   @Override
   public SourceReader<T, DummySplit> createReader(SourceReaderContext sourceReaderContext) {
-    return new SourceReader<T, DummySplit>() {
+    return new SourceReader<>() {
       @Override
       public void start() {
         // Do nothing
       }
 
+      @SuppressWarnings("unchecked")
       @Override
       public InputStatus pollNext(ReaderOutput<T> output) {
         Tuple2<T, Long> next = (Tuple2<T, Long>) QUEUES.get(index).poll();
 
         if (next != null) {
           if (next.f0 == null) {
-            // No more input
-            return InputStatus.END_OF_INPUT;
-          }
-
-          if (next.f1 == null) {
+            if (next.f1 == null) {
+              // No more input
+              return InputStatus.END_OF_INPUT;
+            } else {
+              output.emitWatermark(new Watermark(next.f1));
+            }
+          } else if (next.f1 == null) {
             // No event time set
             output.collect(next.f0);
           } else {
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
new file mode 100644
index 000000000000..cffcc4eb0471
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
@@ -0,0 +1,378 @@
+/*
+ * 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.operator;
+
+import static org.apache.iceberg.flink.maintenance.operator.ConstantsForTests.DUMMY_NAME;
+import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.FAILED_TASK_COUNTER;
+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.util.Collection;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.api.connector.sink2.CommitterInitContext;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.SupportsCommitter;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
+import org.apache.flink.streaming.api.connector.sink2.SupportsPostCommitTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.Collector;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+@Timeout(value = 10)
+class TestLockRemover extends OperatorTestBase {
+  private static final String[] TASKS = new String[] {"task0", "task1"};
+  private static final TriggerLockFactory.Lock LOCK = new TestingLock();
+  private static final TriggerLockFactory.Lock RECOVERY_LOCK = new TestingLock();
+
+  @TempDir private File checkpointDir;
+
+  @BeforeEach
+  void before() {
+    MetricsReporterFactoryForTests.reset();
+  }
+
+  @Test
+  void testProcess() throws Exception {
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
+    source
+        .dataStream()
+        .transform(
+            DUMMY_NAME,
+            TypeInformation.of(Void.class),
+            new LockRemover(new TestingLockFactory(), Lists.newArrayList(TASKS)))
+        .setParallelism(1);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+
+      LOCK.tryLock();
+      assertThat(LOCK.isHeld()).isTrue();
+
+      // Start a successful trigger for task1 and assert the return value is correct
+      processAndCheck(source, new TaskResult(0, 0L, true, Lists.newArrayList()));
+
+      // Assert that the lock is removed
+      assertThat(LOCK.isHeld()).isFalse();
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  @Test
+  void testInSink() throws Exception {
+    String sinkName = "TestSink";
+    Configuration config = new Configuration();
+    config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
+    config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file://" + checkpointDir.getPath());
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+    env.enableCheckpointing(10);
+    ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
+    source.dataStream().global().sinkTo(new SinkTest()).name(sinkName).setParallelism(1);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+
+      LOCK.tryLock();
+      assertThat(LOCK.isHeld()).isTrue();
+
+      // Start a successful trigger for task1 and assert the return value is correct
+      processAndCheck(source, new TaskResult(0, 0L, true, Lists.newArrayList()), sinkName + ": ");
+
+      // Assert that the lock is removed
+      assertThat(LOCK.isHeld()).isFalse();
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  @Test
+  void testMetrics() throws Exception {
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
+    source
+        .dataStream()
+        .transform(
+            DUMMY_NAME,
+            TypeInformation.of(Void.class),
+            new LockRemover(new TestingLockFactory(), Lists.newArrayList(TASKS)))
+        .setParallelism(1);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+      // Start the 2 successful and one failed result trigger for task1, and 3 successful for task2
+      processAndCheck(source, new TaskResult(0, 0L, true, Lists.newArrayList()));
+      processAndCheck(source, new TaskResult(1, 1L, true, Lists.newArrayList()));
+      processAndCheck(source, new TaskResult(1, 2L, true, Lists.newArrayList()));
+      processAndCheck(source, new TaskResult(0, 3L, false, Lists.newArrayList()));
+      processAndCheck(source, new TaskResult(0, 4L, true, Lists.newArrayList()));
+      processAndCheck(source, new TaskResult(1, 5L, true, Lists.newArrayList()));
+
+      Awaitility.await()
+          .until(
+              () ->
+                  MetricsReporterFactoryForTests.counter(
+                          DUMMY_NAME + "." + TASKS[1] + "." + SUCCEEDED_TASK_COUNTER)
+                      .equals(3L));
+
+      // Final check all the counters
+      MetricsReporterFactoryForTests.assertCounters(
+          new ImmutableMap.Builder<String, Long>()
+              .put(DUMMY_NAME + "." + TASKS[0] + "." + SUCCEEDED_TASK_COUNTER, 2L)
+              .put(DUMMY_NAME + "." + TASKS[0] + "." + FAILED_TASK_COUNTER, 1L)
+              .put(DUMMY_NAME + "." + TASKS[1] + "." + SUCCEEDED_TASK_COUNTER, 3L)
+              .put(DUMMY_NAME + "." + TASKS[1] + "." + FAILED_TASK_COUNTER, 0L)
+              .build());
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  /**
+   * The test checks if the recovery watermark is only removed if the watermark has arrived from
+   * both upstream sources.
+   *
+   * @throws Exception if any
+   */
+  @Test
+  void testRecovery() throws Exception {
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    ManualSource<TaskResult> source1 =
+        new ManualSource<>(env, TypeInformation.of(TaskResult.class));
+    ManualSource<TaskResult> source2 =
+        new ManualSource<>(env, TypeInformation.of(TaskResult.class));
+    source1
+        .dataStream()
+        .union(source2.dataStream())
+        .transform(
+            DUMMY_NAME,
+            TypeInformation.of(Void.class),
+            new LockRemover(new TestingLockFactory(), Lists.newArrayList(TASKS[0])))
+        .setParallelism(1);
+
+    JobClient jobClient = null;
+    try {
+      jobClient = env.executeAsync();
+      RECOVERY_LOCK.tryLock();
+      assertThat(RECOVERY_LOCK.isHeld()).isTrue();
+
+      processAndCheck(source1, new TaskResult(0, 0L, true, Lists.newArrayList()));
+
+      source1.sendRecord(new TaskResult(0, 1L, true, Lists.newArrayList()));
+      // we receive the second result - this will not happen in real use cases, but with this we can
+      // be sure that the previous watermark is processed
+      Awaitility.await()
+          .until(
+              () ->
+                  MetricsReporterFactoryForTests.counter(
+                          DUMMY_NAME + "." + TASKS[0] + "." + SUCCEEDED_TASK_COUNTER)
+                      .equals(2L));
+
+      // We did not remove the recovery lock, as no watermark received from the other source
+      assertThat(RECOVERY_LOCK.isHeld()).isTrue();
+
+      // Recovery arrives
+      source1.sendWatermark(10L);
+      source2.sendWatermark(10L);
+
+      Awaitility.await().until(() -> !RECOVERY_LOCK.isHeld());
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  private void processAndCheck(ManualSource<TaskResult> source, TaskResult input) {
+    processAndCheck(source, input, null);
+  }
+
+  private void processAndCheck(
+      ManualSource<TaskResult> source, TaskResult input, String counterPrefix) {
+    source.sendRecord(input);
+    source.sendWatermark(input.startEpoch());
+
+    String counterName =
+        (counterPrefix != null ? counterPrefix : "")
+            .concat(
+                input.success()
+                    ? DUMMY_NAME + "." + TASKS[input.taskIndex()] + "." + SUCCEEDED_TASK_COUNTER
+                    : DUMMY_NAME + "." + TASKS[input.taskIndex()] + "." + FAILED_TASK_COUNTER);
+    Long counterValue = MetricsReporterFactoryForTests.counter(counterName);
+    Long expected = counterValue != null ? counterValue + 1 : 1L;
+
+    Awaitility.await()
+        .until(() -> expected.equals(MetricsReporterFactoryForTests.counter(counterName)));
+  }
+
+  private static class TestingLockFactory implements TriggerLockFactory {
+    @Override
+    public void open() {
+      // Do nothing
+    }
+
+    @Override
+    public Lock createLock() {
+      return LOCK;
+    }
+
+    @Override
+    public Lock createRecoveryLock() {
+      return RECOVERY_LOCK;
+    }
+
+    @Override
+    public void close() {
+      // Do nothing
+    }
+  }
+
+  private static class TestingLock implements TriggerLockFactory.Lock {
+    private boolean locked = false;
+
+    @Override
+    public boolean tryLock() {
+      if (isHeld()) {
+        return false;
+      } else {
+        locked = true;
+        return true;
+      }
+    }
+
+    @Override
+    public boolean isHeld() {
+      return locked;
+    }
+
+    @Override
+    public void unlock() {
+      locked = false;
+    }
+  }
+
+  private static class SinkTest
+      implements Sink<TaskResult>,
+          SupportsCommitter<TaskResult>,
+          SupportsPostCommitTopology<TaskResult> {
+    @Override
+    public SinkWriter<TaskResult> createWriter(InitContext initContext) {
+      return new CommittingSinkWriter<TaskResult, TaskResult>() {
+        private final Collection<TaskResult> received = Lists.newArrayList();
+
+        @Override
+        public Collection<TaskResult> prepareCommit() {
+          Collection<TaskResult> result = Lists.newArrayList(received);
+          received.clear();
+          return result;
+        }
+
+        @Override
+        public void write(TaskResult taskResult, Context context) {
+          received.add(taskResult);
+        }
+
+        @Override
+        public void flush(boolean b) {
+          // noop
+        }
+
+        @Override
+        public void close() {
+          // noop
+        }
+      };
+    }
+
+    @Override
+    public Committer<TaskResult> createCommitter(CommitterInitContext committerInitContext) {
+      return new Committer<>() {
+        @Override
+        public void commit(Collection<CommitRequest<TaskResult>> collection) {
+          // noop
+        }
+
+        @Override
+        public void close() {
+          // noop
+        }
+      };
+    }
+
+    @Override
+    public SimpleVersionedSerializer<TaskResult> getCommittableSerializer() {
+      return new SimpleVersionedSerializer<>() {
+        @Override
+        public int getVersion() {
+          return 0;
+        }
+
+        @Override
+        public byte[] serialize(TaskResult taskResult) {
+          return new byte[0];
+        }
+
+        @Override
+        public TaskResult deserialize(int i, byte[] bytes) {
+          return null;
+        }
+      };
+    }
+
+    @Override
+    public void addPostCommitTopology(DataStream<CommittableMessage<TaskResult>> committables) {
+      committables
+          .flatMap(
+              new FlatMapFunction<CommittableMessage<TaskResult>, TaskResult>() {
+                @Override
+                public void flatMap(
+                    CommittableMessage<TaskResult> taskResultCommittableMessage,
+                    Collector<TaskResult> collector) {
+                  if (taskResultCommittableMessage instanceof CommittableWithLineage) {
+                    collector.collect(
+                        ((CommittableWithLineage<TaskResult>) taskResultCommittableMessage)
+                            .getCommittable());
+                  }
+                }
+              })
+          .transform(
+              DUMMY_NAME,
+              TypeInformation.of(Void.class),
+              new LockRemover(new TestingLockFactory(), Lists.newArrayList(TASKS[0])));
+    }
+  }
+}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
index ccb90ec33d9c..cffcc4eb0471 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemover.java
@@ -24,7 +24,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
-import java.time.Duration;
 import java.util.Collection;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -100,8 +99,8 @@ void testInSink() throws Exception {
     Configuration config = new Configuration();
     config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
     config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file://" + checkpointDir.getPath());
-    config.set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofMillis(10));
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+    env.enableCheckpointing(10);
     ManualSource<TaskResult> source = new ManualSource<>(env, TypeInformation.of(TaskResult.class));
     source.dataStream().global().sinkTo(new SinkTest()).name(sinkName).setParallelism(1);