Flink: Backport DynamicCommitter jobId fix to v1.20 and v2.0 (#16648)

Backport commit #16011.

Author: lrsb

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
@@ -111,6 +112,17 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
     }
 
     /*
+      Group the incoming commit requests into a three-level structure before committing:
+
+        Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest>>>>
+              |              |                  |              |
+              |              |                  |              +-- commit requests at that checkpoint
+              |              |                  +-- checkpointId, sorted ascending so older commits go first
+              |              +-- (jobId, operatorId) of the producing aggregator; deduplication against the
+              |                  table's snapshot summaries is per (jobId, operatorId), so each group is
+              |                  walked against the ancestor chain independently
+              +-- (tableName, branch); we load the table and its ancestor chain once per TableKey
+
       Each (table, branch, checkpoint) triplet must have only one commit request.
       There may be commit requests from previous checkpoints which have not been committed yet.
 
@@ -119,43 +131,58 @@ public void commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
       DynamicWriteResultAggregator. Iceberg 1.12 will remove this, and users should upgrade to the 1.11 release first
       to migrate their state to a single commit request per checkpoint.
     */
-    Map<TableKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>> commitRequestMap =
-        Maps.newHashMap();
+    Map<TableKey, Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+        commitRequestMap = Maps.newHashMap();
     for (CommitRequest<DynamicCommittable> request : commitRequests) {
-      NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> committables =
-          commitRequestMap.computeIfAbsent(
-              new TableKey(request.getCommittable()), unused -> Maps.newTreeMap());
-      committables
-          .computeIfAbsent(request.getCommittable().checkpointId(), unused -> Lists.newArrayList())
+      DynamicCommittable committable = request.getCommittable();
+      commitRequestMap
+          .computeIfAbsent(committable.key(), unused -> Maps.newHashMap())
+          .computeIfAbsent(new JobOperatorKey(committable), unused -> Maps.newTreeMap())
+          .computeIfAbsent(committable.checkpointId(), unused -> Lists.newArrayList())
           .add(request);
     }
 
-    for (Map.Entry<TableKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>> entry :
-        commitRequestMap.entrySet()) {
-      Table table = catalog.loadTable(TableIdentifier.parse(entry.getKey().tableName()));
-      DynamicCommittable last = entry.getValue().lastEntry().getValue().get(0).getCommittable();
-      Snapshot latestSnapshot = table.snapshot(entry.getKey().branch());
+    for (Map.Entry<
+            TableKey,
+            Map<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+        tableEntry : commitRequestMap.entrySet()) {
+      TableKey tableKey = tableEntry.getKey();
+      Table table = catalog.loadTable(TableIdentifier.parse(tableKey.tableName()));
+      Snapshot latestSnapshot = table.snapshot(tableKey.branch());
       Iterable<Snapshot> ancestors =
           latestSnapshot != null
               ? SnapshotUtil.ancestorsOf(latestSnapshot.snapshotId(), table::snapshot)
               : List.of();
-      long maxCommittedCheckpointId =
-          getMaxCommittedCheckpointId(ancestors, last.jobId(), last.operatorId());
-
-      NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> skippedCommitRequests =
-          entry.getValue().headMap(maxCommittedCheckpointId, true);
-      LOG.debug(
-          "Skipping {} commit requests: {}", skippedCommitRequests.size(), skippedCommitRequests);
-      // Mark the already committed FilesCommittable(s) as finished
-      skippedCommitRequests
-          .values()
-          .forEach(list -> list.forEach(CommitRequest::signalAlreadyCommitted));
-
-      NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> uncommitted =
-          entry.getValue().tailMap(maxCommittedCheckpointId, false);
-      if (!uncommitted.isEmpty()) {
-        commitPendingRequests(
-            table, entry.getKey().branch(), uncommitted, last.jobId(), last.operatorId());
+
+      List<Map.Entry<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>>
+          jobEntries = Lists.newArrayList(tableEntry.getValue().entrySet());
+      // Preserve checkpoint order across groups so that older-jobId commits land before newer-jobId
+      // ones when the batch mixes committables from different jobIds (e.g. state replay after a
+      // restart). Within a (jobId, operatorId) group, checkpoint order is already guaranteed by
+      // the inner NavigableMap.
+      jobEntries.sort(Comparator.comparingLong(entry -> entry.getValue().firstKey()));
+
+      for (Map.Entry<JobOperatorKey, NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>>
+          jobEntry : jobEntries) {
+        JobOperatorKey jobKey = jobEntry.getKey();
+        long maxCommittedCheckpointId =
+            getMaxCommittedCheckpointId(ancestors, jobKey.jobId(), jobKey.operatorId());
+
+        NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> skippedCommitRequests =
+            jobEntry.getValue().headMap(maxCommittedCheckpointId, true);
+        LOG.debug(
+            "Skipping {} commit requests: {}", skippedCommitRequests.size(), skippedCommitRequests);
+        // Mark the already committed FilesCommittable(s) as finished
+        skippedCommitRequests
+            .values()
+            .forEach(list -> list.forEach(CommitRequest::signalAlreadyCommitted));
+
+        NavigableMap<Long, List<CommitRequest<DynamicCommittable>>> uncommitted =
+            jobEntry.getValue().tailMap(maxCommittedCheckpointId, false);
+        if (!uncommitted.isEmpty()) {
+          commitPendingRequests(
+              table, tableKey.branch(), uncommitted, jobKey.jobId(), jobKey.operatorId());
+        }
       }
     }
   }

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java

@@ -0,0 +1,57 @@
+/*
+ * 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.sink.dynamic;
+
+import java.util.Objects;
+
+class JobOperatorKey {
+  private final String jobId;
+  private final String operatorId;
+
+  JobOperatorKey(DynamicCommittable committable) {
+    this.jobId = committable.jobId();
+    this.operatorId = committable.operatorId();
+  }
+
+  String jobId() {
+    return jobId;
+  }
+
+  String operatorId() {
+    return operatorId;
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    }
+
+    if (!(other instanceof JobOperatorKey that)) {
+      return false;
+    }
+
+    return Objects.equals(jobId, that.jobId) && Objects.equals(operatorId, that.operatorId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(jobId, operatorId);
+  }
+}

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableKey.java

@@ -33,11 +33,6 @@ class TableKey {
     this.branch = branch;
   }
 
-  TableKey(DynamicCommittable committable) {
-    this.tableName = committable.key().tableName();
-    this.branch = committable.key().branch();
-  }
-
   String tableName() {
     return tableName;
   }