Flink: Backport PR #16324 to v2.0 and v1.20 (#16338)

Agent-Logs-Url: https://github.com/kevinjqliu/iceberg/sessions/682ce8b4-890f-41a9-a89a-b1f2873be44c

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: kevinjqliu <9057843+kevinjqliu@users.noreply.github.com>

Author: kevinjqliu

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java

@@ -66,6 +66,7 @@ public void open(OpenContext openContext) throws Exception {
   public void processElement(Trigger trigger, Context ctx, Collector<String> collector)
       throws Exception {
     try {
+      table.refresh();
       table
           .snapshots()
           .forEach(

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestListMetadataFiles.java

@@ -23,8 +23,10 @@
 import java.util.List;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.junit.jupiter.api.Test;
 
 class TestListMetadataFiles extends OperatorTestBase {
@@ -87,4 +89,37 @@ void testMetadataFilesWithEmptyTable() throws Exception {
       assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
     }
   }
+
+  @Test
+  void testMetadataFilesIncludesSnapshotsAddedAfterOpen() throws Exception {
+    Table table = createTable();
+    insert(table, 1, "a");
+
+    try (OneInputStreamOperatorTestHarness<Trigger, String> testHarness =
+        ProcessFunctionTestHarnesses.forProcessFunction(
+            new ListMetadataFiles(OperatorTestBase.DUMMY_TABLE_NAME, 0, tableLoader()))) {
+      testHarness.open();
+
+      // Add more snapshots AFTER the operator has been opened
+      insert(table, 2, "b");
+      insert(table, 3, "c");
+
+      OperatorTestBase.trigger(testHarness);
+
+      List<String> tableMetadataFiles = testHarness.extractOutputValues();
+
+      // Verify that manifest lists from ALL 3 snapshots are present, not just the first one.
+      // Without table.refresh() in processElement, only snapshot 1's files would be emitted.
+      table.refresh();
+      List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
+      assertThat(snapshots).hasSize(3);
+      for (Snapshot snapshot : snapshots) {
+        assertThat(tableMetadataFiles).contains(snapshot.manifestListLocation());
+      }
+      // Verify total count matches what 3 snapshots should produce
+      assertThat(tableMetadataFiles).hasSize(24);
+
+      assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+    }
+  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java

@@ -66,6 +66,7 @@ public void open(OpenContext openContext) throws Exception {
   public void processElement(Trigger trigger, Context ctx, Collector<String> collector)
       throws Exception {
     try {
+      table.refresh();
       table
           .snapshots()
           .forEach(

flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestListMetadataFiles.java

@@ -23,8 +23,10 @@
 import java.util.List;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.junit.jupiter.api.Test;
 
 class TestListMetadataFiles extends OperatorTestBase {
@@ -87,4 +89,37 @@ void testMetadataFilesWithEmptyTable() throws Exception {
       assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
     }
   }
+
+  @Test
+  void testMetadataFilesIncludesSnapshotsAddedAfterOpen() throws Exception {
+    Table table = createTable();
+    insert(table, 1, "a");
+
+    try (OneInputStreamOperatorTestHarness<Trigger, String> testHarness =
+        ProcessFunctionTestHarnesses.forProcessFunction(
+            new ListMetadataFiles(OperatorTestBase.DUMMY_TABLE_NAME, 0, tableLoader()))) {
+      testHarness.open();
+
+      // Add more snapshots AFTER the operator has been opened
+      insert(table, 2, "b");
+      insert(table, 3, "c");
+
+      OperatorTestBase.trigger(testHarness);
+
+      List<String> tableMetadataFiles = testHarness.extractOutputValues();
+
+      // Verify that manifest lists from ALL 3 snapshots are present, not just the first one.
+      // Without table.refresh() in processElement, only snapshot 1's files would be emitted.
+      table.refresh();
+      List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
+      assertThat(snapshots).hasSize(3);
+      for (Snapshot snapshot : snapshots) {
+        assertThat(tableMetadataFiles).contains(snapshot.manifestListLocation());
+      }
+      // Verify total count matches what 3 snapshots should produce
+      assertThat(tableMetadataFiles).hasSize(24);
+
+      assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+    }
+  }
 }