[AMORO-4162] Upgrade Iceberg dependency from 1.6.1 to 1.7.2

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataRecordFactory.java

@@ -29,10 +29,12 @@
 class RowDataRecordFactory implements RecordFactory<RowData> {
   private final RowType rowType;
   private final TypeSerializer[] fieldSerializers;
+  private final RowData.FieldGetter[] fieldGetters;
 
   RowDataRecordFactory(RowType rowType) {
     this.rowType = rowType;
     this.fieldSerializers = createFieldSerializers(rowType);
+    this.fieldGetters = createFieldGetters(rowType);
   }
 
   static TypeSerializer[] createFieldSerializers(RowType rowType) {
@@ -41,6 +43,14 @@ static TypeSerializer[] createFieldSerializers(RowType rowType) {
         .toArray(TypeSerializer[]::new);
   }
 
+  static RowData.FieldGetter[] createFieldGetters(RowType rowType) {
+    RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[rowType.getFieldCount()];
+    for (int i = 0; i < rowType.getFieldCount(); i++) {
+      fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+    }
+    return fieldGetters;
+  }
+
   @Override
   public RowData[] createBatch(int batchSize) {
     RowData[] arr = new RowData[batchSize];
@@ -57,6 +67,7 @@ public void clone(RowData from, RowData[] batch, int position) {
     // Clone method will allocate a new GenericRowData object
     // if the target object is NOT a GenericRowData.
     // So we should always set the clone return value back to the array.
-    batch[position] = RowDataUtil.clone(from, batch[position], rowType, fieldSerializers);
+    batch[position] =
+        RowDataUtil.clone(from, batch[position], rowType, fieldSerializers, fieldGetters);
   }
 }

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/IcebergClassUtil.java

@@ -35,12 +35,12 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
 import org.apache.iceberg.flink.sink.TaskWriterFactory;
 import org.apache.iceberg.flink.source.FlinkInputFormat;
 import org.apache.iceberg.flink.source.ScanContext;
 import org.apache.iceberg.flink.source.StreamingReaderOperator;
 import org.apache.iceberg.io.FileIO;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.util.ThreadPools;
 
 import java.lang.reflect.Constructor;
@@ -76,7 +76,7 @@ public static KeySelector<RowData, Object> newPartitionKeySelector(
     }
   }
 
-  public static OneInputStreamOperator<WriteResult, Void> newIcebergFilesCommitter(
+  public static OneInputStreamOperator<FlinkWriteResult, Void> newIcebergFilesCommitter(
       TableLoader tableLoader, boolean replacePartitions, String branch, PartitionSpec spec) {
     try {
       Class<?> clazz = forName(ICEBERG_FILE_COMMITTER_CLASS);
@@ -89,7 +89,7 @@ public static OneInputStreamOperator<WriteResult, Void> newIcebergFilesCommitter
               String.class,
               PartitionSpec.class);
       c.setAccessible(true);
-      return (OneInputStreamOperator<WriteResult, Void>)
+      return (OneInputStreamOperator<FlinkWriteResult, Void>)
           c.newInstance(
               tableLoader,
               replacePartitions,
@@ -105,13 +105,13 @@ public static OneInputStreamOperator<WriteResult, Void> newIcebergFilesCommitter
     }
   }
 
-  public static OneInputStreamOperator<WriteResult, Void> newIcebergFilesCommitter(
+  public static OneInputStreamOperator<FlinkWriteResult, Void> newIcebergFilesCommitter(
       TableLoader tableLoader,
       boolean replacePartitions,
       String branch,
       PartitionSpec spec,
       AuthenticatedFileIO authenticatedFileIO) {
-    OneInputStreamOperator<WriteResult, Void> obj =
+    OneInputStreamOperator<FlinkWriteResult, Void> obj =
         newIcebergFilesCommitter(tableLoader, replacePartitions, branch, spec);
     return (OneInputStreamOperator) ProxyUtil.getProxy(obj, authenticatedFileIO);
   }

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/FlinkSink.java

@@ -65,8 +65,8 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
 import org.apache.iceberg.flink.sink.TaskWriterFactory;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.util.PropertyUtil;
 import org.slf4j.Logger;
@@ -153,15 +153,15 @@ DataStreamSink<?> withEmit(
         DataStream<RowData> input,
         MixedFormatLogWriter logWriter,
         MixedFormatFileWriter fileWriter,
-        OneInputStreamOperator<WriteResult, Void> committer,
+        OneInputStreamOperator<FlinkWriteResult, Void> committer,
         int writeOperatorParallelism,
         MetricsGenerator metricsGenerator,
         String emitMode) {
       SingleOutputStreamOperator writerStream =
           input
               .transform(
                   MixedFormatWriter.class.getName(),
-                  TypeExtractor.createTypeInfo(WriteResult.class),
+                  TypeExtractor.createTypeInfo(FlinkWriteResult.class),
                   new MixedFormatWriter<>(logWriter, fileWriter, metricsGenerator))
               .name(String.format("MixedFormatWriter %s(%s)", table.name(), emitMode))
               .setParallelism(writeOperatorParallelism);
@@ -414,7 +414,7 @@ private static TaskWriterFactory<RowData> createTaskWriterFactory(
     return new MixedFormatRowDataTaskWriterFactory(mixedTable, flinkSchema, overwrite);
   }
 
-  public static OneInputStreamOperator<WriteResult, Void> createFileCommitter(
+  public static OneInputStreamOperator<FlinkWriteResult, Void> createFileCommitter(
       MixedTable mixedTable,
       MixedFormatTableLoader tableLoader,
       boolean overwrite,
@@ -424,7 +424,7 @@ public static OneInputStreamOperator<WriteResult, Void> createFileCommitter(
         mixedTable, tableLoader, overwrite, branch, spec, MIXED_FORMAT_EMIT_FILE);
   }
 
-  public static OneInputStreamOperator<WriteResult, Void> createFileCommitter(
+  public static OneInputStreamOperator<FlinkWriteResult, Void> createFileCommitter(
       MixedTable mixedTable,
       MixedFormatTableLoader tableLoader,
       boolean overwrite,

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatFileWriter.java

@@ -34,6 +34,7 @@
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
 import org.apache.iceberg.flink.sink.TaskWriterFactory;
 import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;
@@ -46,8 +47,8 @@
 import java.util.stream.IntStream;
 
 /** This is mixed-format table includes writing file data to un keyed table and keyed table. */
-public class MixedFormatFileWriter extends AbstractStreamOperator<WriteResult>
-    implements OneInputStreamOperator<RowData, WriteResult>, BoundedOneInput {
+public class MixedFormatFileWriter extends AbstractStreamOperator<FlinkWriteResult>
+    implements OneInputStreamOperator<RowData, FlinkWriteResult>, BoundedOneInput {
 
   private static final long serialVersionUID = 1L;
   private static final Logger LOG = LoggerFactory.getLogger(MixedFormatFileWriter.class);
@@ -61,6 +62,7 @@ public class MixedFormatFileWriter extends AbstractStreamOperator<WriteResult>
 
   private transient TaskWriter<RowData> writer;
   private transient int subTaskId;
+  private transient long currentCheckpointId;
   private transient int attemptId;
   /**
    * Load table in runtime, because that table's refresh method will be invoked in serialization.
@@ -140,6 +142,7 @@ private long getMask(int subTaskId) {
 
   @Override
   public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+    this.currentCheckpointId = checkpointId;
     table
         .io()
         .doAs(
@@ -153,6 +156,7 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
 
   @Override
   public void endInput() throws Exception {
+    this.currentCheckpointId = Long.MAX_VALUE;
     table
         .io()
         .doAs(
@@ -205,7 +209,7 @@ private void emit(WriteResult writeResult) {
     if (shouldEmit(writeResult)) {
       // Only emit a non-empty WriteResult to committer operator, thus avoiding submitting too much
       // empty snapshots.
-      output.collect(new StreamRecord<>(writeResult));
+      output.collect(new StreamRecord<>(new FlinkWriteResult(currentCheckpointId, writeResult)));
     }
   }
 

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/FlinkTestBase.java

@@ -64,7 +64,6 @@
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
-import org.apache.iceberg.flink.MiniClusterResource;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/MiniClusterResource.java

@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.flink;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+/** Forked from org.apache.iceberg.flink.MiniClusterResource which was removed in Iceberg 1.7.0. */
+public class MiniClusterResource {
+
+  private static final int DEFAULT_TM_NUM = 1;
+  private static final int DEFAULT_PARALLELISM = 4;
+
+  public static final Configuration DISABLE_CLASSLOADER_CHECK_CONFIG =
+      new Configuration().set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+  private MiniClusterResource() {}
+
+  public static MiniClusterWithClientResource createWithClassloaderCheckDisabled() {
+    return new MiniClusterWithClientResource(
+        new MiniClusterResourceConfiguration.Builder()
+            .setNumberTaskManagers(DEFAULT_TM_NUM)
+            .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+            .setConfiguration(DISABLE_CLASSLOADER_CHECK_CONFIG)
+            .build());
+  }
+}

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java

@@ -33,6 +33,7 @@
 import org.apache.amoro.TableTestHelper;
 import org.apache.amoro.catalog.BasicCatalogTestHelper;
 import org.apache.amoro.catalog.CatalogTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
 import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
@@ -51,7 +52,6 @@
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.CollectionUtil;
-import org.apache.iceberg.flink.MiniClusterResource;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/AmoroCatalogITCaseBase.java

@@ -21,6 +21,7 @@
 import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED;
 
 import org.apache.amoro.TestAms;
+import org.apache.amoro.flink.MiniClusterResource;
 import org.apache.amoro.formats.AmoroCatalogTestBase;
 import org.apache.amoro.formats.AmoroCatalogTestHelper;
 import org.apache.amoro.hive.TestHMS;
@@ -36,7 +37,6 @@
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
 import org.junit.ClassRule;
 
 import java.io.IOException;

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/CatalogITCaseBase.java

@@ -23,6 +23,7 @@
 import org.apache.amoro.TableTestHelper;
 import org.apache.amoro.catalog.CatalogTestHelper;
 import org.apache.amoro.catalog.TableTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.state.StateBackend;
@@ -35,7 +36,6 @@
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
 import org.junit.ClassRule;
 
 public abstract class CatalogITCaseBase extends TableTestBase {

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyedOverwrite.java

@@ -24,6 +24,7 @@
 import org.apache.amoro.catalog.BasicCatalogTestHelper;
 import org.apache.amoro.catalog.CatalogTestHelper;
 import org.apache.amoro.flink.FlinkTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
 import org.apache.amoro.flink.util.DataUtil;
 import org.apache.amoro.hive.TestHMS;
 import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
@@ -32,7 +33,6 @@
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.ClassRule;

amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestAutomaticLogWriter.java

@@ -60,7 +60,7 @@
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -256,12 +256,12 @@ public void testHasNotCaughtUp() throws Exception {
     }
 
     List<Object[]> expects = new LinkedList<>();
-    List<WriteResult> results;
+    List<FlinkWriteResult> results;
     testKeyedTable.refresh();
     Assert.assertFalse(
         Boolean.parseBoolean(
             testKeyedTable.properties().getOrDefault(LOG_STORE_CATCH_UP.key(), "false")));
-    try (TestOneInputStreamOperatorIntern<RowData, WriteResult> harness =
+    try (TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> harness =
         createSingleProducer(1, jobId, topic, gap)) {
       DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
       expects.add(
@@ -314,7 +314,7 @@ public void testHasNotCaughtUp() throws Exception {
 
     // check expects accuracy.
     Assert.assertEquals(3, results.size());
-    results.forEach(result -> Assert.assertEquals(1, result.dataFiles().length));
+    results.forEach(result -> Assert.assertEquals(1, result.writeResult().dataFiles().length));
     List<Object[]> expected = isGapNone ? expects : expects.subList(2, expects.size());
     checkLogstoreDataAccuracy(topic, expected);
     testKeyedTable.refresh();
@@ -350,7 +350,7 @@ private void checkLogstoreDataAccuracy(String topic, List<Object[]> expects) {
             .collect(Collectors.toList()));
   }
 
-  public TestOneInputStreamOperatorIntern<RowData, WriteResult> createSingleProducer(
+  public TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> createSingleProducer(
       int maxParallelism, byte[] jobId, String topic, Duration writeLogstoreWatermarkGap)
       throws Exception {
     return createProducer(
@@ -364,7 +364,7 @@ public TestOneInputStreamOperatorIntern<RowData, WriteResult> createSingleProduc
         writeLogstoreWatermarkGap);
   }
 
-  private TestOneInputStreamOperatorIntern<RowData, WriteResult> createProducer(
+  private TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> createProducer(
       int maxParallelism,
       int parallelism,
       int subTaskId,
@@ -402,10 +402,10 @@ private TestOneInputStreamOperatorIntern<RowData, WriteResult> createProducer(
             (RowType) FLINK_SCHEMA.toRowDataType().getLogicalType(),
             tableLoader);
 
-    MixedFormatWriter<WriteResult> mixedFormatWriter =
+    MixedFormatWriter<FlinkWriteResult> mixedFormatWriter =
         new MixedFormatWriter<>(automaticLogWriter, streamWriter, metricsGenerator);
 
-    TestOneInputStreamOperatorIntern<RowData, WriteResult> harness =
+    TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> harness =
         new TestOneInputStreamOperatorIntern<>(
             mixedFormatWriter,
             maxParallelism,