Repository: apache/iceberg (language: python)

Issue:
Flink: Maintenance - MonitorSource
### Feature Request / Improvement

The responsibility of the Monitor source is to periodically check the table metadata and based on the new commits, generate the TableChange messages for the Trigger Manager.

### Query engine

None


Discussion / hints:

Fix patch:
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java
new file mode 100644
index 000000000000..d74b2349b1de
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java
@@ -0,0 +1,206 @@
+/*
+ * 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.io.IOException;
+import java.util.Iterator;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimitedSourceReader;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiter;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.iceberg.DataOperations;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Monitors an Iceberg table for changes */
+@Internal
+public class MonitorSource extends SingleThreadedIteratorSource<TableChange> {
+  private static final Logger LOG = LoggerFactory.getLogger(MonitorSource.class);
+
+  private final TableLoader tableLoader;
+  private final RateLimiterStrategy rateLimiterStrategy;
+  private final long maxReadBack;
+
+  /**
+   * Creates a {@link org.apache.flink.api.connector.source.Source} which monitors an Iceberg table
+   * for changes.
+   *
+   * @param tableLoader used for accessing the table
+   * @param rateLimiterStrategy limits the frequency the table is checked
+   * @param maxReadBack sets the number of snapshots read before stopping change collection
+   */
+  public MonitorSource(
+      TableLoader tableLoader, RateLimiterStrategy rateLimiterStrategy, long maxReadBack) {
+    Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
+    Preconditions.checkNotNull(rateLimiterStrategy, "Rate limiter strategy should no be null");
+    Preconditions.checkArgument(maxReadBack > 0, "Need to read at least 1 snapshot to work");
+
+    this.tableLoader = tableLoader;
+    this.rateLimiterStrategy = rateLimiterStrategy;
+    this.maxReadBack = maxReadBack;
+  }
+
+  @Override
+  public Boundedness getBoundedness() {
+    return Boundedness.CONTINUOUS_UNBOUNDED;
+  }
+
+  @Override
+  public TypeInformation<TableChange> getProducedType() {
+    return TypeInformation.of(TableChange.class);
+  }
+
+  @Override
+  Iterator<TableChange> createIterator() {
+    return new TableChangeIterator(tableLoader, null, maxReadBack);
+  }
+
+  @Override
+  SimpleVersionedSerializer<Iterator<TableChange>> iteratorSerializer() {
+    return new TableChangeIteratorSerializer(tableLoader, maxReadBack);
+  }
+
+  @Override
+  public SourceReader<TableChange, GlobalSplit<TableChange>> createReader(
+      SourceReaderContext readerContext) throws Exception {
+    RateLimiter rateLimiter = rateLimiterStrategy.createRateLimiter(1);
+    return new RateLimitedSourceReader<>(super.createReader(readerContext), rateLimiter);
+  }
+
+  /** The Iterator which returns the latest changes on an Iceberg table. */
+  @VisibleForTesting
+  static class TableChangeIterator implements Iterator<TableChange> {
+    private Long lastSnapshotId;
+    private final long maxReadBack;
+    private final Table table;
+
+    TableChangeIterator(TableLoader tableLoader, Long lastSnapshotId, long maxReadBack) {
+      this.lastSnapshotId = lastSnapshotId;
+      this.maxReadBack = maxReadBack;
+      tableLoader.open();
+      this.table = tableLoader.loadTable();
+    }
+
+    @Override
+    public boolean hasNext() {
+      return true;
+    }
+
+    @Override
+    public TableChange next() {
+      try {
+        table.refresh();
+        Snapshot currentSnapshot = table.currentSnapshot();
+        Long current = currentSnapshot != null ? currentSnapshot.snapshotId() : null;
+        Long checking = current;
+        TableChange event = TableChange.empty();
+        long readBack = 0;
+        while (checking != null && !checking.equals(lastSnapshotId) && ++readBack <= maxReadBack) {
+          Snapshot snapshot = table.snapshot(checking);
+          if (snapshot != null) {
+            if (!DataOperations.REPLACE.equals(snapshot.operation())) {
+              LOG.debug("Reading snapshot {}", snapshot.snapshotId());
+              event.merge(new TableChange(snapshot, table.io()));
+            } else {
+              LOG.debug("Skipping replace snapshot {}", snapshot.snapshotId());
+            }
+
+            checking = snapshot.parentId();
+          } else {
+            // If the last snapshot has been removed from the history
+            checking = null;
+          }
+        }
+
+        lastSnapshotId = current;
+        return event;
+      } catch (Exception e) {
+        LOG.warn("Failed to fetch table changes for {}", table, e);
+        return TableChange.empty();
+      }
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("lastSnapshotId", lastSnapshotId)
+          .add("maxReadBack", maxReadBack)
+          .add("table", table)
+          .toString();
+    }
+  }
+
+  private static final class TableChangeIteratorSerializer
+      implements SimpleVersionedSerializer<Iterator<TableChange>> {
+
+    private static final int CURRENT_VERSION = 1;
+    private final TableLoader tableLoader;
+    private final long maxReadBack;
+
+    TableChangeIteratorSerializer(TableLoader tableLoader, long maxReadBack) {
+      this.tableLoader = tableLoader;
+      this.maxReadBack = maxReadBack;
+    }
+
+    @Override
+    public int getVersion() {
+      return CURRENT_VERSION;
+    }
+
+    @Override
+    public byte[] serialize(Iterator<TableChange> iterator) throws IOException {
+      Preconditions.checkArgument(
+          iterator instanceof TableChangeIterator,
+          "Use TableChangeIterator iterator. Found incompatible type: %s",
+          iterator.getClass());
+
+      TableChangeIterator tableChangeIterator = (TableChangeIterator) iterator;
+      DataOutputSerializer out = new DataOutputSerializer(8);
+      long toStore =
+          tableChangeIterator.lastSnapshotId != null ? tableChangeIterator.lastSnapshotId : -1L;
+      out.writeLong(toStore);
+      return out.getCopyOfBuffer();
+    }
+
+    @Override
+    public TableChangeIterator deserialize(int version, byte[] serialized) throws IOException {
+      if (version == CURRENT_VERSION) {
+        DataInputDeserializer in = new DataInputDeserializer(serialized);
+        long fromStore = in.readLong();
+        return new TableChangeIterator(
+            tableLoader, fromStore != -1 ? fromStore : null, maxReadBack);
+      } else {
+        throw new IOException("Unrecognized version or corrupt state: " + version);
+      }
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/SingleThreadedIteratorSource.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/SingleThreadedIteratorSource.java
new file mode 100644
index 000000000000..20c7684d9700
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/SingleThreadedIteratorSource.java
@@ -0,0 +1,197 @@
+/*
+ * 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.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.connector.source.lib.util.IteratorSourceEnumerator;
+import org.apache.flink.api.connector.source.lib.util.IteratorSourceReader;
+import org.apache.flink.api.connector.source.lib.util.IteratorSourceSplit;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * Implementation of the Source V2 API which uses an iterator to read the elements, and uses a
+ * single thread to do so.
+ *
+ * @param <T> The return type of the source
+ */
+@Internal
+public abstract class SingleThreadedIteratorSource<T>
+    implements Source<
+            T,
+            SingleThreadedIteratorSource.GlobalSplit<T>,
+            Collection<SingleThreadedIteratorSource.GlobalSplit<T>>>,
+        ResultTypeQueryable<T> {
+  private static final String PARALLELISM_ERROR = "Parallelism should be set to 1";
+
+  /**
+   * Creates the iterator to return the elements which then emitted by the source.
+   *
+   * @return iterator for the elements
+   */
+  abstract Iterator<T> createIterator();
+
+  /**
+   * Serializes the iterator, which is used to save and restore the state of the source.
+   *
+   * @return serializer for the iterator
+   */
+  abstract SimpleVersionedSerializer<Iterator<T>> iteratorSerializer();
+
+  @Override
+  public SplitEnumerator<GlobalSplit<T>, Collection<GlobalSplit<T>>> createEnumerator(
+      SplitEnumeratorContext<GlobalSplit<T>> enumContext) {
+    Preconditions.checkArgument(enumContext.currentParallelism() == 1, PARALLELISM_ERROR);
+    return new IteratorSourceEnumerator<>(
+        enumContext, ImmutableList.of(new GlobalSplit<>(createIterator())));
+  }
+
+  @Override
+  public SplitEnumerator<GlobalSplit<T>, Collection<GlobalSplit<T>>> restoreEnumerator(
+      SplitEnumeratorContext<GlobalSplit<T>> enumContext, Collection<GlobalSplit<T>> checkpoint) {
+    Preconditions.checkArgument(enumContext.currentParallelism() == 1, PARALLELISM_ERROR);
+    return new IteratorSourceEnumerator<>(enumContext, checkpoint);
+  }
+
+  @Override
+  public SimpleVersionedSerializer<GlobalSplit<T>> getSplitSerializer() {
+    return new SplitSerializer<>(iteratorSerializer());
+  }
+
+  @Override
+  public SimpleVersionedSerializer<Collection<GlobalSplit<T>>> getEnumeratorCheckpointSerializer() {
+    return new EnumeratorSerializer<>(iteratorSerializer());
+  }
+
+  @Override
+  public SourceReader<T, GlobalSplit<T>> createReader(SourceReaderContext readerContext)
+      throws Exception {
+    Preconditions.checkArgument(readerContext.getIndexOfSubtask() == 0, PARALLELISM_ERROR);
+    return new IteratorSourceReader<>(readerContext);
+  }
+
+  /** The single split of the {@link SingleThreadedIteratorSource}. */
+  static class GlobalSplit<T> implements IteratorSourceSplit<T, Iterator<T>> {
+    private final Iterator<T> iterator;
+
+    GlobalSplit(Iterator<T> iterator) {
+      this.iterator = iterator;
+    }
+
+    @Override
+    public String splitId() {
+      return "1";
+    }
+
+    @Override
+    public Iterator<T> getIterator() {
+      return iterator;
+    }
+
+    @Override
+    public IteratorSourceSplit<T, Iterator<T>> getUpdatedSplitForIterator(
+        final Iterator<T> newIterator) {
+      return new GlobalSplit<>(newIterator);
+    }
+
+    @Override
+    public String toString() {
+      return String.format("GlobalSplit (%s)", iterator);
+    }
+  }
+
+  private static final class SplitSerializer<T>
+      implements SimpleVersionedSerializer<GlobalSplit<T>> {
+    private final SimpleVersionedSerializer<Iterator<T>> iteratorSerializer;
+
+    SplitSerializer(SimpleVersionedSerializer<Iterator<T>> iteratorSerializer) {
+      this.iteratorSerializer = iteratorSerializer;
+    }
+
+    private static final int CURRENT_VERSION = 1;
+
+    @Override
+    public int getVersion() {
+      return CURRENT_VERSION;
+    }
+
+    @Override
+    public byte[] serialize(GlobalSplit<T> split) throws IOException {
+      return iteratorSerializer.serialize(split.iterator);
+    }
+
+    @Override
+    public GlobalSplit<T> deserialize(int version, byte[] serialized) throws IOException {
+      return new GlobalSplit<>(iteratorSerializer.deserialize(version, serialized));
+    }
+  }
+
+  private static final class EnumeratorSerializer<T>
+      implements SimpleVersionedSerializer<Collection<GlobalSplit<T>>> {
+    private static final int CURRENT_VERSION = 1;
+    private final SimpleVersionedSerializer<Iterator<T>> iteratorSerializer;
+
+    EnumeratorSerializer(SimpleVersionedSerializer<Iterator<T>> iteratorSerializer) {
+      this.iteratorSerializer = iteratorSerializer;
+    }
+
+    @Override
+    public int getVersion() {
+      return CURRENT_VERSION;
+    }
+
+    @Override
+    public byte[] serialize(Collection<GlobalSplit<T>> checkpoint) throws IOException {
+      Preconditions.checkArgument(checkpoint.size() < 2, PARALLELISM_ERROR);
+      if (checkpoint.isEmpty()) {
+        return new byte[] {0};
+      } else {
+        byte[] iterator = iteratorSerializer.serialize(checkpoint.iterator().next().getIterator());
+        byte[] result = new byte[iterator.length + 1];
+        result[0] = 1;
+        System.arraycopy(iterator, 0, result, 1, iterator.length);
+        return result;
+      }
+    }
+
+    @Override
+    public Collection<GlobalSplit<T>> deserialize(int version, byte[] serialized)
+        throws IOException {
+      if (serialized[0] == 0) {
+        return Lists.newArrayList();
+      } else {
+        byte[] iterator = new byte[serialized.length - 1];
+        System.arraycopy(serialized, 1, iterator, 0, serialized.length - 1);
+        return Lists.newArrayList(
+            new GlobalSplit<>(iteratorSerializer.deserialize(version, iterator)));
+      }
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java
new file mode 100644
index 000000000000..452ed80ed0e5
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java
@@ -0,0 +1,133 @@
+/*
+ * 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.Objects;
+import org.apache.flink.annotation.Internal;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+
+/** Event describing changes in an Iceberg table */
+@Internal
+class TableChange {
+  private int dataFileNum;
+  private int deleteFileNum;
+  private long dataFileSize;
+  private long deleteFileSize;
+  private int commitNum;
+
+  TableChange(
+      int dataFileNum, int deleteFileNum, long dataFileSize, long deleteFileSize, int commitNum) {
+    this.dataFileNum = dataFileNum;
+    this.deleteFileNum = deleteFileNum;
+    this.dataFileSize = dataFileSize;
+    this.deleteFileSize = deleteFileSize;
+    this.commitNum = commitNum;
+  }
+
+  TableChange(Snapshot snapshot, FileIO io) {
+    Iterable<DataFile> dataFiles = snapshot.addedDataFiles(io);
+    Iterable<DeleteFile> deleteFiles = snapshot.addedDeleteFiles(io);
+
+    dataFiles.forEach(
+        dataFile -> {
+          this.dataFileNum++;
+          this.dataFileSize += dataFile.fileSizeInBytes();
+        });
+
+    deleteFiles.forEach(
+        deleteFile -> {
+          this.deleteFileNum++;
+          this.deleteFileSize += deleteFile.fileSizeInBytes();
+        });
+
+    this.commitNum = 1;
+  }
+
+  static TableChange empty() {
+    return new TableChange(0, 0, 0L, 0L, 0);
+  }
+
+  int dataFileNum() {
+    return dataFileNum;
+  }
+
+  int deleteFileNum() {
+    return deleteFileNum;
+  }
+
+  long dataFileSize() {
+    return dataFileSize;
+  }
+
+  long deleteFileSize() {
+    return deleteFileSize;
+  }
+
+  public int commitNum() {
+    return commitNum;
+  }
+
+  public void merge(TableChange other) {
+    this.dataFileNum += other.dataFileNum;
+    this.deleteFileNum += other.deleteFileNum;
+    this.dataFileSize += other.dataFileSize;
+    this.deleteFileSize += other.deleteFileSize;
+    this.commitNum += other.commitNum;
+  }
+
+  TableChange copy() {
+    return new TableChange(dataFileNum, deleteFileNum, dataFileSize, deleteFileSize, commitNum);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("dataFileNum", dataFileNum)
+        .add("deleteFileNum", deleteFileNum)
+        .add("dataFileSize", dataFileSize)
+        .add("deleteFileSize", deleteFileSize)
+        .add("commitNum", commitNum)
+        .toString();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    } else if (other == null || getClass() != other.getClass()) {
+      return false;
+    }
+
+    TableChange that = (TableChange) other;
+    return dataFileNum == that.dataFileNum
+        && deleteFileNum == that.deleteFileNum
+        && dataFileSize == that.dataFileSize
+        && deleteFileSize == that.deleteFileSize
+        && commitNum == that.commitNum;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(dataFileNum, deleteFileNum, dataFileSize, deleteFileSize, commitNum);
+  }
+}


Test patch:
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/CollectingSink.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/CollectingSink.java
new file mode 100644
index 000000000000..a49459d61a54
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/CollectingSink.java
@@ -0,0 +1,115 @@
+/*
+ * 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.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/** Sink for collecting output during testing. */
+class CollectingSink<T> implements Sink<T> {
+  private static final long serialVersionUID = 1L;
+  private static final List<BlockingQueue<Object>> queues =
+      Collections.synchronizedList(Lists.newArrayListWithExpectedSize(1));
+  private static final AtomicInteger numSinks = new AtomicInteger(-1);
+  private final int index;
+
+  /** Creates a new sink which collects the elements received. */
+  CollectingSink() {
+    this.index = numSinks.incrementAndGet();
+    queues.add(new LinkedBlockingQueue<>());
+  }
+
+  /**
+   * Gets all the remaining output received by this {@link Sink}.
+   *
+   * @return all the remaining output
+   */
+  List<T> remainingOutput() {
+    return Lists.newArrayList((BlockingQueue<T>) queues.get(this.index));
+  }
+
+  /**
+   * Check if there is no remaining output received by this {@link Sink}.
+   *
+   * @return <code>true</code> if there is no remaining output
+   */
+  boolean isEmpty() {
+    return queues.get(this.index).isEmpty();
+  }
+
+  /**
+   * Wait until the next element received by the {@link Sink}.
+   *
+   * @param timeout for the poll
+   * @return The first element received by this {@link Sink}
+   * @throws TimeoutException if no element received until the timeout
+   */
+  T poll(Duration timeout) throws TimeoutException {
+    Object element;
+
+    try {
+      element = queues.get(this.index).poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    } catch (InterruptedException var4) {
+      throw new RuntimeException(var4);
+    }
+
+    if (element == null) {
+      throw new TimeoutException();
+    } else {
+      return (T) element;
+    }
+  }
+
+  @Override
+  public SinkWriter<T> createWriter(InitContext context) {
+    return new CollectingWriter<>(index);
+  }
+
+  private static class CollectingWriter<T> implements SinkWriter<T> {
+    private final int index;
+
+    CollectingWriter(int index) {
+      this.index = index;
+    }
+
+    @Override
+    public void write(T element, Context context) {
+      queues.get(index).add(element);
+    }
+
+    @Override
+    public void flush(boolean endOfInput) {
+      // Nothing to do here
+    }
+
+    @Override
+    public void close() {
+      // Nothing to do here
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkSqlExtension.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkSqlExtension.java
new file mode 100644
index 000000000000..90790b373d5f
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkSqlExtension.java
@@ -0,0 +1,132 @@
+/*
+ * 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.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.CatalogLoader;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+/**
+ * Junit 5 extension for running Flink SQL queries. {@link
+ * org.apache.flink.test.junit5.MiniClusterExtension} is used for executing the SQL batch jobs.
+ */
+public class FlinkSqlExtension implements BeforeEachCallback, AfterEachCallback {
+  private final String catalogName;
+  private final Map<String, String> catalogProperties;
+  private final String databaseName;
+  private final Path warehouse;
+  private final CatalogLoader catalogLoader;
+  private TableEnvironment tableEnvironment;
+
+  public FlinkSqlExtension(
+      String catalogName, Map<String, String> catalogProperties, String databaseName) {
+    this.catalogName = catalogName;
+    this.catalogProperties = Maps.newHashMap(catalogProperties);
+    this.databaseName = databaseName;
+
+    // Add temporary dir as a warehouse location
+    try {
+      this.warehouse = Files.createTempDirectory("warehouse");
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    this.catalogProperties.put(
+        CatalogProperties.WAREHOUSE_LOCATION, String.format("file://%s", warehouse));
+    this.catalogLoader =
+        CatalogLoader.hadoop(catalogName, new Configuration(), this.catalogProperties);
+  }
+
+  @Override
+  public void beforeEach(ExtensionContext context) {
+    // We need to recreate the tableEnvironment for every test as the minicluster is recreated
+    this.tableEnvironment =
+        TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+    exec("CREATE CATALOG %s WITH %s", catalogName, toWithClause(catalogProperties));
+    exec("CREATE DATABASE %s.%s", catalogName, databaseName);
+    exec("USE CATALOG %s", catalogName);
+    exec("USE %s", databaseName);
+  }
+
+  @Override
+  public void afterEach(ExtensionContext context) throws IOException {
+    List<Row> tables = exec("SHOW TABLES");
+    tables.forEach(t -> exec("DROP TABLE IF EXISTS %s", t.getField(0)));
+    exec("USE CATALOG default_catalog");
+    exec("DROP CATALOG IF EXISTS %s", catalogName);
+    Files.walk(warehouse).sorted(Comparator.reverseOrder()).map(Path::toFile).forEach(File::delete);
+  }
+
+  /**
+   * Executes an SQL query with the given parameters. The parameter substitution is done by {@link
+   * String#format(String, Object...)}.
+   *
+   * @param query to run
+   * @param parameters to substitute to the query
+   * @return The {@link Row}s returned by the query
+   */
+  public List<Row> exec(String query, Object... parameters) {
+    TableResult tableResult = tableEnvironment.executeSql(String.format(query, parameters));
+    try (CloseableIterator<Row> iter = tableResult.collect()) {
+      return Lists.newArrayList(iter);
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to collect table result", e);
+    }
+  }
+
+  /**
+   * Returns the {@link TableLoader} which could be used to access the given table.
+   *
+   * @param tableName of the table
+   * @return the {@link TableLoader} for the table
+   */
+  public TableLoader tableLoader(String tableName) {
+    TableLoader tableLoader =
+        TableLoader.fromCatalog(catalogLoader, TableIdentifier.of(databaseName, tableName));
+    tableLoader.open();
+    return tableLoader;
+  }
+
+  private static String toWithClause(Map<String, String> props) {
+    return String.format(
+        "(%s)",
+        props.entrySet().stream()
+            .map(e -> String.format("'%s'='%s'", e.getKey(), e.getValue()))
+            .collect(Collectors.joining(",")));
+  }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkStreamingTestUtils.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkStreamingTestUtils.java
new file mode 100644
index 000000000000..9cdc55cb0cce
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/FlinkStreamingTestUtils.java
@@ -0,0 +1,73 @@
+/*
+ * 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.io.File;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
+import org.awaitility.Awaitility;
+
+class FlinkStreamingTestUtils {
+  private FlinkStreamingTestUtils() {
+    // Do not instantiate
+  }
+
+  /**
+   * Close the {@link JobClient} and wait for the job closure. If the savepointDir is specified, it
+   * stops the job with a savepoint.
+   *
+   * @param jobClient the job to close
+   * @param savepointDir the savepointDir to store the last savepoint. If <code>null</code> then
+   *     stop without a savepoint.
+   * @return configuration for restarting the job from the savepoint
+   */
+  static Configuration closeJobClient(JobClient jobClient, File savepointDir) {
+    Configuration conf = new Configuration();
+    if (jobClient != null) {
+      if (savepointDir != null) {
+        // Stop with savepoint
+        jobClient.stopWithSavepoint(false, savepointDir.getPath(), SavepointFormatType.CANONICAL);
+        // Wait until the savepoint is created and the job has been stopped
+        Awaitility.await().until(() -> savepointDir.listFiles(File::isDirectory).length == 1);
+        conf.set(
+            SavepointConfigOptions.SAVEPOINT_PATH,
+            savepointDir.listFiles(File::isDirectory)[0].getAbsolutePath());
+      } else {
+        jobClient.cancel();
+      }
+
+      // Wait until the job has been stopped
+      Awaitility.await().until(() -> jobClient.getJobStatus().get().isTerminalState());
+      return conf;
+    }
+
+    return null;
+  }
+
+  /**
+   * Close the {@link JobClient} and wait for the job closure.
+   *
+   * @param jobClient the job to close
+   */
+  static void closeJobClient(JobClient jobClient) {
+    closeJobClient(jobClient, null);
+  }
+}
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
new file mode 100644
index 000000000000..e08742a89ddb
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/ManualSource.java
@@ -0,0 +1,316 @@
+/*
+ * 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.ArrayDeque;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Queues;
+import org.jetbrains.annotations.Nullable;
+
+/** Testing source implementation for Flink sources which can be triggered manually. */
+class ManualSource<T>
+    implements Source<T, ManualSource.DummySplit, ManualSource.DummyCheckpoint>,
+        ResultTypeQueryable<T> {
+
+  private static final long serialVersionUID = 1L;
+  private static final List<ArrayDeque<Tuple2<?, Long>>> queues =
+      Collections.synchronizedList(Lists.newArrayList());
+  private static final List<CompletableFuture<Void>> availabilities =
+      Collections.synchronizedList(Lists.newArrayList());
+  private static int numSources = 0;
+  private final TypeInformation<T> type;
+  private final int index;
+  private transient DataStream<T> stream;
+  private final transient StreamExecutionEnvironment env;
+
+  /**
+   * Creates a new source for testing.
+   *
+   * @param env to register the source
+   * @param type of the events returned by the source
+   */
+  ManualSource(StreamExecutionEnvironment env, TypeInformation<T> type) {
+    this.type = type;
+    this.env = env;
+    this.index = numSources++;
+    queues.add(Queues.newArrayDeque());
+    availabilities.add(new CompletableFuture<>());
+  }
+
+  /**
+   * Emit a new record from the source.
+   *
+   * @param event to emit
+   */
+  void sendRecord(T event) {
+    this.sendInternal(Tuple2.of(event, null));
+  }
+
+  /**
+   * Emit a new record with the given event time from the source.
+   *
+   * @param event to emit
+   * @param eventTime of the event
+   */
+  void sendRecord(T event, long eventTime) {
+    this.sendInternal(Tuple2.of(event, eventTime));
+  }
+
+  /**
+   * Emit a watermark from the source.
+   *
+   * @param timeStamp of the watermark
+   */
+  void sendWatermark(long timeStamp) {
+    this.sendInternal(Tuple2.of(null, timeStamp));
+  }
+
+  /** Mark the source as finished. */
+  void markFinished() {
+    this.sendWatermark(Long.MAX_VALUE);
+    this.sendInternal(Tuple2.of(null, null));
+  }
+
+  /**
+   * Get the {@link DataStream} for this source.
+   *
+   * @return the stream emitted by this source
+   */
+  DataStream<T> dataStream() {
+    if (this.stream == null) {
+      this.stream =
+          this.env
+              .fromSource(this, WatermarkStrategy.noWatermarks(), "ManualSource-" + index, type)
+              .forceNonParallel();
+    }
+
+    return this.stream;
+  }
+
+  private void sendInternal(Tuple2<?, Long> tuple) {
+    queues.get(index).offer(tuple);
+    availabilities.get(index).complete(null);
+  }
+
+  @Override
+  public Boundedness getBoundedness() {
+    return Boundedness.CONTINUOUS_UNBOUNDED;
+  }
+
+  @Override
+  public SplitEnumerator<DummySplit, DummyCheckpoint> createEnumerator(
+      SplitEnumeratorContext<DummySplit> enumContext) {
+    return new DummyCheckpointEnumerator();
+  }
+
+  @Override
+  public SplitEnumerator<DummySplit, DummyCheckpoint> restoreEnumerator(
+      SplitEnumeratorContext<DummySplit> enumContext, DummyCheckpoint checkpoint) {
+    return new DummyCheckpointEnumerator();
+  }
+
+  @Override
+  public SimpleVersionedSerializer<DummySplit> getSplitSerializer() {
+    return new NoOpDummySplitSerializer();
+  }
+
+  @Override
+  public SimpleVersionedSerializer<DummyCheckpoint> getEnumeratorCheckpointSerializer() {
+    return new NoOpDummyCheckpointSerializer();
+  }
+
+  @Override
+  public SourceReader<T, DummySplit> createReader(SourceReaderContext sourceReaderContext) {
+    return new SourceReader<T, DummySplit>() {
+      @Override
+      public void start() {
+        // Do nothing
+      }
+
+      @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) {
+            // No event time set
+            output.collect(next.f0);
+          } else {
+            // With event time
+            output.collect(next.f0, next.f1);
+          }
+        }
+
+        availabilities.set(index, new CompletableFuture<>());
+        return queues.get(index).isEmpty()
+            ? InputStatus.NOTHING_AVAILABLE
+            : InputStatus.MORE_AVAILABLE;
+      }
+
+      @Override
+      public List<DummySplit> snapshotState(long checkpointId) {
+        return Lists.newArrayList(new DummySplit());
+      }
+
+      @Override
+      public CompletableFuture<Void> isAvailable() {
+        return availabilities.get(index);
+      }
+
+      @Override
+      public void addSplits(List<DummySplit> splits) {
+        // do nothing
+      }
+
+      @Override
+      public void notifyNoMoreSplits() {
+        // do nothing
+      }
+
+      @Override
+      public void close() {
+        // do nothing
+      }
+    };
+  }
+
+  @Override
+  public TypeInformation<T> getProducedType() {
+    return this.type;
+  }
+
+  /**
+   * Placeholder because the ManualSource itself implicitly represents the only split and does not
+   * require an actual split object.
+   */
+  public static class DummySplit implements SourceSplit {
+    @Override
+    public String splitId() {
+      return "dummy";
+    }
+  }
+
+  /**
+   * Placeholder because the ManualSource does not support fault-tolerance and thus does not require
+   * actual checkpointing.
+   */
+  public static class DummyCheckpoint {}
+
+  /** Placeholder because the ManualSource does not need enumeration, but checkpointing needs it. */
+  private static class DummyCheckpointEnumerator
+      implements SplitEnumerator<DummySplit, DummyCheckpoint> {
+
+    @Override
+    public void start() {
+      // do nothing
+    }
+
+    @Override
+    public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
+      // do nothing
+    }
+
+    @Override
+    public void addSplitsBack(List<DummySplit> splits, int subtaskId) {
+      // do nothing
+    }
+
+    @Override
+    public void addReader(int subtaskId) {
+      // do nothing
+    }
+
+    @Override
+    public DummyCheckpoint snapshotState(long checkpointId) {
+      return new DummyCheckpoint();
+    }
+
+    @Override
+    public void close() {
+      // do nothing
+    }
+  }
+
+  /**
+   * Not used - only required to avoid NullPointerException. The split is not transferred from the
+   * enumerator, it is implicitly represented by the ManualSource.
+   */
+  private static class NoOpDummySplitSerializer implements SimpleVersionedSerializer<DummySplit> {
+    @Override
+    public int getVersion() {
+      return 0;
+    }
+
+    @Override
+    public byte[] serialize(DummySplit split) {
+      return new byte[0];
+    }
+
+    @Override
+    public DummySplit deserialize(int version, byte[] serialized) {
+      return new DummySplit();
+    }
+  }
+
+  /**
+   * Not used - only required to avoid NullPointerException. The split is not transferred from the
+   * enumerator, it is implicitly represented by the ManualSource.
+   */
+  private static class NoOpDummyCheckpointSerializer
+      implements SimpleVersionedSerializer<DummyCheckpoint> {
+    @Override
+    public int getVersion() {
+      return 0;
+    }
+
+    @Override
+    public byte[] serialize(DummyCheckpoint split) {
+      return new byte[0];
+    }
+
+    @Override
+    public DummyCheckpoint deserialize(int version, byte[] serialized) {
+      return new DummyCheckpoint();
+    }
+  }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
new file mode 100644
index 000000000000..269ae681b02e
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -0,0 +1,51 @@
+/*
+ * 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.MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class OperatorTestBase {
+  private static final int NUMBER_TASK_MANAGERS = 1;
+  private static final int SLOTS_PER_TASK_MANAGER = 8;
+
+  static final String TABLE_NAME = "test_table";
+
+  @RegisterExtension
+  protected static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+      new MiniClusterExtension(
+          new MiniClusterResourceConfiguration.Builder()
+              .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
+              .setNumberSlotsPerTaskManager(SLOTS_PER_TASK_MANAGER)
+              .setConfiguration(new Configuration(DISABLE_CLASSLOADER_CHECK_CONFIG))
+              .build());
+
+  @RegisterExtension
+  final FlinkSqlExtension sql =
+      new FlinkSqlExtension(
+          "catalog",
+          ImmutableMap.of("type", "iceberg", FlinkCatalogFactory.ICEBERG_CATALOG_TYPE, "hadoop"),
+          "db");
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
new file mode 100644
index 000000000000..876d64214560
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
@@ -0,0 +1,362 @@
+/*
+ * 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.FlinkStreamingTestUtils.closeJobClient;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.awaitility.Awaitility;
+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;
+
+class TestMonitorSource extends OperatorTestBase {
+  private static final TableChange EMPTY_EVENT = TableChange.empty();
+  private static final RateLimiterStrategy HIGH_RATE = RateLimiterStrategy.perSecond(100.0);
+  private static final RateLimiterStrategy LOW_RATE = RateLimiterStrategy.perSecond(1.0 / 10000.0);
+
+  @TempDir private File checkpointDir;
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  void testChangeReaderIterator(boolean withDelete) {
+    if (withDelete) {
+      sql.exec(
+          "CREATE TABLE %s (id int, data varchar, PRIMARY KEY(`id`) NOT ENFORCED) WITH ('format-version'='2', 'write.upsert.enabled'='true')",
+          TABLE_NAME);
+    } else {
+      sql.exec("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
+    }
+
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+    Table table = tableLoader.loadTable();
+
+    MonitorSource.TableChangeIterator iterator =
+        new MonitorSource.TableChangeIterator(tableLoader, null, Long.MAX_VALUE);
+
+    // For an empty table we get an empty result
+    assertThat(iterator.next()).isEqualTo(EMPTY_EVENT);
+
+    // Add a single commit and get back the commit data in the event
+    sql.exec("INSERT INTO %s VALUES (1, 'a')", TABLE_NAME);
+    table.refresh();
+    TableChange expected = tableChangeWithLastSnapshot(table, TableChange.empty());
+    assertThat(iterator.next()).isEqualTo(expected);
+    // Make sure that consecutive calls do not return the data again
+    assertThat(iterator.next()).isEqualTo(EMPTY_EVENT);
+
+    // Add two more commits, but fetch the data in one loop
+    sql.exec("INSERT INTO %s VALUES (2, 'b')", TABLE_NAME);
+    table.refresh();
+    expected = tableChangeWithLastSnapshot(table, TableChange.empty());
+
+    sql.exec("INSERT INTO %s VALUES (3, 'c')", TABLE_NAME);
+    table.refresh();
+    expected = tableChangeWithLastSnapshot(table, expected);
+
+    assertThat(iterator.next()).isEqualTo(expected);
+    // Make sure that consecutive calls do not return the data again
+    assertThat(iterator.next()).isEqualTo(EMPTY_EVENT);
+  }
+
+  /**
+   * Create a table and check that the source returns the data as new commits arrive to the table.
+   */
+  @Test
+  void testSource() throws Exception {
+    sql.exec(
+        "CREATE TABLE %s (id int, data varchar) "
+            + "WITH ('flink.max-continuous-empty-commits'='100000')",
+        TABLE_NAME);
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+    Table table = tableLoader.loadTable();
+    DataStream<TableChange> events =
+        env.fromSource(
+                new MonitorSource(tableLoader, HIGH_RATE, Long.MAX_VALUE),
+                WatermarkStrategy.noWatermarks(),
+                "TableChangeSource")
+            .forceNonParallel();
+
+    // Sink to collect the results
+    CollectingSink<TableChange> result = new CollectingSink<>();
+    events.sinkTo(result);
+
+    JobClient jobClient = null;
+    try {
+      // First result is an empty event
+      jobClient = env.executeAsync("Table Change Source Test");
+      assertThat(result.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+
+      // Insert some data
+      File dataDir = new File(new Path(table.location(), "data").toUri().getPath());
+      dataDir.mkdir();
+      GenericAppenderHelper dataAppender =
+          new GenericAppenderHelper(table, FileFormat.PARQUET, dataDir.toPath());
+      List<Record> batch1 = RandomGenericData.generate(table.schema(), 2, 1);
+      dataAppender.appendToTable(batch1);
+
+      // Wait until the changes are committed
+      Awaitility.await()
+          .until(
+              () -> {
+                table.refresh();
+                return table.currentSnapshot() != null;
+              });
+
+      table.refresh();
+      long size = firstFileLength(table);
+
+      // Wait until the first non-empty event has arrived, and check the expected result
+      Awaitility.await()
+          .until(
+              () -> {
+                TableChange newEvent = result.poll(Duration.ofSeconds(5L));
+                // Fetch every empty event from the beginning
+                while (newEvent.equals(EMPTY_EVENT)) {
+                  newEvent = result.poll(Duration.ofSeconds(5L));
+                }
+
+                // The first non-empty event should contain the expected value
+                return newEvent.equals(new TableChange(1, 0, size, 0L, 1));
+              });
+    } finally {
+      closeJobClient(jobClient);
+    }
+  }
+
+  /** Check that the {@link MonitorSource} operator state is restored correctly. */
+  @Test
+  void testStateRestore(@TempDir File savepointDir) throws Exception {
+    sql.exec("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
+    sql.exec("INSERT INTO %s VALUES (1, 'a')", TABLE_NAME);
+
+    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(1000);
+
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+    DataStream<TableChange> events =
+        env.fromSource(
+                new MonitorSource(tableLoader, HIGH_RATE, Long.MAX_VALUE),
+                WatermarkStrategy.noWatermarks(),
+                "TableChangeSource")
+            .forceNonParallel();
+
+    // Sink to collect the results
+    CollectingSink<TableChange> result = new CollectingSink<>();
+    events.sinkTo(result);
+
+    // Start the job
+    Configuration conf;
+    JobClient jobClient = null;
+    AtomicReference<TableChange> firstNonEmptyEvent = new AtomicReference<>();
+    try {
+      jobClient = env.executeAsync("Table Change Source Test");
+
+      Awaitility.await()
+          .until(
+              () -> {
+                TableChange newEvent = result.poll(Duration.ofSeconds(5L));
+                // Fetch every empty event from the beginning
+                while (newEvent.equals(EMPTY_EVENT)) {
+                  newEvent = result.poll(Duration.ofSeconds(5L));
+                }
+
+                // The first non-empty event should contain the expected value
+                firstNonEmptyEvent.set(newEvent);
+                return true;
+              });
+    } finally {
+      // Stop with savepoint
+      conf = closeJobClient(jobClient, savepointDir);
+    }
+
+    // Restore from savepoint, create the same topology with a different env
+    env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
+    events =
+        env.fromSource(
+                new MonitorSource(tableLoader, LOW_RATE, Long.MAX_VALUE),
+                WatermarkStrategy.noWatermarks(),
+                "TableChangeSource")
+            .forceNonParallel();
+    CollectingSink<TableChange> resultWithSavepoint = new CollectingSink<>();
+    events.sinkTo(resultWithSavepoint);
+
+    // Make sure that the job with restored source does not read new records from the table
+    JobClient clientWithSavepoint = null;
+    try {
+      clientWithSavepoint = env.executeAsync("Table Change Source test with savepoint");
+
+      assertThat(resultWithSavepoint.poll(Duration.ofSeconds(5L))).isEqualTo(EMPTY_EVENT);
+    } finally {
+      closeJobClient(clientWithSavepoint, null);
+    }
+
+    // Restore without savepoint
+    env = StreamExecutionEnvironment.getExecutionEnvironment();
+    events =
+        env.fromSource(
+                new MonitorSource(tableLoader, LOW_RATE, Long.MAX_VALUE),
+                WatermarkStrategy.noWatermarks(),
+                "TableChangeSource")
+            .forceNonParallel();
+    CollectingSink<TableChange> resultWithoutSavepoint = new CollectingSink<>();
+    events.sinkTo(resultWithoutSavepoint);
+
+    // Make sure that a new job without state reads the event as expected
+    JobClient clientWithoutSavepoint = null;
+    try {
+      clientWithoutSavepoint = env.executeAsync("Table Change Source Test without savepoint");
+      assertThat(resultWithoutSavepoint.poll(Duration.ofSeconds(5L)))
+          .isEqualTo(firstNonEmptyEvent.get());
+    } finally {
+      closeJobClient(clientWithoutSavepoint);
+    }
+  }
+
+  @Test
+  void testNotOneParallelismThrows() {
+    sql.exec("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
+
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+
+    env.fromSource(
+            new MonitorSource(tableLoader, HIGH_RATE, Long.MAX_VALUE),
+            WatermarkStrategy.noWatermarks(),
+            "TableChangeSource")
+        .setParallelism(2)
+        .print();
+
+    assertThatThrownBy(env::execute)
+        .isInstanceOf(JobExecutionException.class)
+        .rootCause()
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Parallelism should be set to 1");
+  }
+
+  @Test
+  void testMaxReadBack() {
+    sql.exec("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
+    sql.exec("INSERT INTO %s VALUES (1, 'a')", TABLE_NAME);
+    sql.exec("INSERT INTO %s VALUES (2, 'b')", TABLE_NAME);
+    sql.exec("INSERT INTO %s VALUES (3, 'c')", TABLE_NAME);
+
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+
+    MonitorSource.TableChangeIterator iterator =
+        new MonitorSource.TableChangeIterator(tableLoader, null, 1);
+
+    // For a single maxReadBack we only get a single change
+    assertThat(iterator.next().commitNum()).isEqualTo(1);
+
+    iterator = new MonitorSource.TableChangeIterator(tableLoader, null, 2);
+
+    // Expecting 2 commits/snapshots for maxReadBack=2
+    assertThat(iterator.next().commitNum()).isEqualTo(2);
+
+    iterator = new MonitorSource.TableChangeIterator(tableLoader, null, Long.MAX_VALUE);
+
+    // For maxReadBack Long.MAX_VALUE we get every change
+    assertThat(iterator.next().commitNum()).isEqualTo(3);
+  }
+
+  @Test
+  void testSkipReplace() {
+    sql.exec("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
+    sql.exec("INSERT INTO %s VALUES (1, 'a')", TABLE_NAME);
+
+    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
+    tableLoader.open();
+
+    MonitorSource.TableChangeIterator iterator =
+        new MonitorSource.TableChangeIterator(tableLoader, null, Long.MAX_VALUE);
+
+    // Read the current snapshot
+    assertThat(iterator.next().commitNum()).isEqualTo(1);
+
+    // Create a DataOperations.REPLACE snapshot
+    Table table = tableLoader.loadTable();
+    DataFile dataFile =
+        table.snapshots().iterator().next().addedDataFiles(table.io()).iterator().next();
+    RewriteFiles rewrite = tableLoader.loadTable().newRewrite();
+    // Replace the file with itself for testing purposes
+    rewrite.deleteFile(dataFile);
+    rewrite.addFile(dataFile);
+    rewrite.commit();
+
+    // Check that the rewrite is ignored
+    assertThat(iterator.next()).isEqualTo(EMPTY_EVENT);
+  }
+
+  private static long firstFileLength(Table table) {
+    return table.currentSnapshot().addedDataFiles(table.io()).iterator().next().fileSizeInBytes();
+  }
+
+  private static TableChange tableChangeWithLastSnapshot(Table table, TableChange previous) {
+    List<DataFile> dataFiles =
+        Lists.newArrayList(table.currentSnapshot().addedDataFiles(table.io()).iterator());
+    List<DeleteFile> deleteFiles =
+        Lists.newArrayList(table.currentSnapshot().addedDeleteFiles(table.io()).iterator());
+
+    long dataSize = dataFiles.stream().mapToLong(d -> d.fileSizeInBytes()).sum();
+    long deleteSize = deleteFiles.stream().mapToLong(d -> d.fileSizeInBytes()).sum();
+    boolean hasDelete = table.currentSnapshot().addedDeleteFiles(table.io()).iterator().hasNext();
+
+    return new TableChange(
+        previous.dataFileNum() + dataFiles.size(),
+        previous.deleteFileNum() + deleteFiles.size(),
+        previous.dataFileSize() + dataSize,
+        previous.deleteFileSize() + deleteSize,
+        previous.commitNum() + 1);
+  }
+}