Flink: Prevent recreation of ManifestOutputFileFactory during flushing

flink/v2.1/build.gradle

@@ -68,9 +68,6 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
 
     implementation libs.datasketches
 
-    // for caching in DynamicSink
-    implementation libs.caffeine
-
     testImplementation libs.flink21.connector.test.utils
     testImplementation libs.flink21.core
     testImplementation libs.flink21.runtime

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java

@@ -74,7 +74,19 @@ public static ManifestOutputFileFactory createOutputFileFactory(
       int subTaskId,
       long attemptNumber) {
     return new ManifestOutputFileFactory(
-        tableSupplier, tableProps, flinkJobId, operatorUniqueId, subTaskId, attemptNumber);
+        tableSupplier, tableProps, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, null);
+  }
+
+  public static ManifestOutputFileFactory createOutputFileFactory(
+      Supplier<Table> tableSupplier,
+      Map<String, String> tableProps,
+      String flinkJobId,
+      String operatorUniqueId,
+      int subTaskId,
+      long attemptNumber,
+      String suffix) {
+    return new ManifestOutputFileFactory(
+        tableSupplier, tableProps, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, suffix);
   }
 
   /**

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java

@@ -22,6 +22,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
+import javax.annotation.Nullable;
 import org.apache.flink.annotation.Internal;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.HasTableOperations;
@@ -43,33 +44,37 @@ public class ManifestOutputFileFactory {
   private final int subTaskId;
   private final long attemptNumber;
   private final AtomicInteger fileCount = new AtomicInteger(0);
+  @Nullable private final String suffix;
 
   ManifestOutputFileFactory(
       Supplier<Table> tableSupplier,
       Map<String, String> props,
       String flinkJobId,
       String operatorUniqueId,
       int subTaskId,
-      long attemptNumber) {
+      long attemptNumber,
+      @Nullable String suffix) {
     this.tableSupplier = tableSupplier;
     this.props = props;
     this.flinkJobId = flinkJobId;
     this.operatorUniqueId = operatorUniqueId;
     this.subTaskId = subTaskId;
     this.attemptNumber = attemptNumber;
+    this.suffix = suffix;
   }
 
   private String generatePath(long checkpointId) {
     return FileFormat.AVRO.addExtension(
         String.format(
             Locale.ROOT,
-            "%s-%s-%05d-%d-%d-%05d",
+            "%s-%s-%05d-%d-%d-%05d%s",
             flinkJobId,
             operatorUniqueId,
             subTaskId,
             attemptNumber,
             checkpointId,
-            fileCount.incrementAndGet()));
+            fileCount.incrementAndGet(),
+            suffix != null ? "-" + suffix : ""));
   }
 
   public OutputFile create(long checkpointId) {

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

@@ -168,7 +168,7 @@ public DataStream<CommittableMessage<DynamicCommittable>> addPreCommitTopology(
         .transform(
             prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"),
             typeInformation,
-            new DynamicWriteResultAggregator(catalogLoader))
+            new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize))
         .uid(prefixIfNotNull(uidPrefix, sinkId + "-pre-commit-topology"));
   }
 

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java

@@ -18,12 +18,10 @@
  */
 package org.apache.iceberg.flink.sink.dynamic;
 
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import java.io.IOException;
-import java.time.Duration;
 import java.util.Collection;
 import java.util.Map;
+import java.util.UUID;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.core.io.SimpleVersionedSerialization;
 import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
@@ -61,23 +59,24 @@ class DynamicWriteResultAggregator
 
   private static final Logger LOG = LoggerFactory.getLogger(DynamicWriteResultAggregator.class);
   private static final byte[] EMPTY_MANIFEST_DATA = new byte[0];
-  private static final Duration CACHE_EXPIRATION_DURATION = Duration.ofMinutes(1);
 
   private final CatalogLoader catalogLoader;
+  private final int cacheMaximumSize;
 
   private long lastCheckpointId = CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1;
 
   private transient Map<WriteTarget, Collection<DynamicWriteResult>> results;
-  private transient Cache<String, Map<Integer, PartitionSpec>> specs;
-  private transient Cache<String, ManifestOutputFileFactory> outputFileFactories;
+  private transient Map<String, Map<Integer, PartitionSpec>> specs;
+  private transient Map<String, ManifestOutputFileFactory> outputFileFactories;
   private transient String flinkJobId;
   private transient String operatorId;
   private transient int subTaskId;
   private transient int attemptId;
   private transient Catalog catalog;
 
-  DynamicWriteResultAggregator(CatalogLoader catalogLoader) {
+  DynamicWriteResultAggregator(CatalogLoader catalogLoader, int cacheMaximumSize) {
     this.catalogLoader = catalogLoader;
+    this.cacheMaximumSize = cacheMaximumSize;
   }
 
   @Override
@@ -94,10 +93,8 @@ public void open() throws Exception {
     this.subTaskId = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
     this.attemptId = getRuntimeContext().getTaskInfo().getAttemptNumber();
     this.results = Maps.newHashMap();
-    this.specs =
-        Caffeine.newBuilder().expireAfterWrite(CACHE_EXPIRATION_DURATION).softValues().build();
-    this.outputFileFactories =
-        Caffeine.newBuilder().expireAfterWrite(CACHE_EXPIRATION_DURATION).softValues().build();
+    this.specs = new LRUCache<>(cacheMaximumSize);
+    this.outputFileFactories = new LRUCache<>(cacheMaximumSize);
     this.catalog = catalogLoader.loadCatalog();
   }
 
@@ -186,18 +183,27 @@ public void processElement(StreamRecord<CommittableMessage<DynamicWriteResult>>
   }
 
   private ManifestOutputFileFactory outputFileFactory(String tableName) {
-    return outputFileFactories.get(
+    return outputFileFactories.computeIfAbsent(
         tableName,
         unused -> {
           Table table = catalog.loadTable(TableIdentifier.parse(tableName));
           specs.put(tableName, table.specs());
+          // Make sure to append an identifier to avoid file clashes in case the factory was to get
+          // re-created during a checkpoint, i.e. due to cache eviction.
+          String fileSuffix = UUID.randomUUID().toString();
           return FlinkManifestUtil.createOutputFileFactory(
-              () -> table, table.properties(), flinkJobId, operatorId, subTaskId, attemptId);
+              () -> table,
+              table.properties(),
+              flinkJobId,
+              operatorId,
+              subTaskId,
+              attemptId,
+              fileSuffix);
         });
   }
 
   private PartitionSpec spec(String tableName, int specId) {
-    Map<Integer, PartitionSpec> knownSpecs = specs.getIfPresent(tableName);
+    Map<Integer, PartitionSpec> knownSpecs = specs.get(tableName);
     if (knownSpecs != null) {
       PartitionSpec spec = knownSpecs.get(specId);
       if (spec != null) {

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java

@@ -210,4 +210,9 @@ private static Set<Integer> getEqualityFields(Table table, Set<Integer> equality
   DynamicWriterMetrics getMetrics() {
     return metrics;
   }
+
+  @VisibleForTesting
+  Map<WriteTarget, RowDataTaskWriterFactory> getTaskWriterFactories() {
+    return taskWriterFactories;
+  }
 }

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkManifest.java

@@ -137,7 +137,7 @@ public void testUserProvidedManifestLocation() throws IOException {
             ManifestOutputFileFactory.FLINK_MANIFEST_LOCATION,
             userProvidedFolder.getAbsolutePath() + "///");
     ManifestOutputFileFactory factory =
-        new ManifestOutputFileFactory(() -> table, props, flinkJobId, operatorId, 1, 1);
+        new ManifestOutputFileFactory(() -> table, props, flinkJobId, operatorId, 1, 1, null);
 
     List<DataFile> dataFiles = generateDataFiles(5);
     DeltaManifests deltaManifests =

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java

@@ -0,0 +1,112 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class TestManifestOutputFileFactory {
+
+  @RegisterExtension
+  static final HadoopCatalogExtension CATALOG_EXTENSION = new HadoopCatalogExtension("db", "table");
+
+  private Table table;
+
+  @BeforeEach
+  void before() throws IOException {
+    table = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), new Schema());
+  }
+
+  @Test
+  public void testFileNameFormat() {
+    String flinkJobId = "job123";
+    String operatorUniqueId = "operator456";
+    int subTaskId = 7;
+    long attemptNumber = 2;
+    long checkpointId = 100;
+    Map<String, String> props = table.properties();
+
+    ManifestOutputFileFactory factory =
+        new ManifestOutputFileFactory(
+            () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, null);
+
+    String file1 = new File(factory.create(checkpointId).location()).getName();
+    assertThat(file1).isEqualTo("job123-operator456-00007-2-100-00001.avro");
+
+    String file2 = new File(factory.create(checkpointId).location()).getName();
+    assertThat(file2).isEqualTo("job123-operator456-00007-2-100-00002.avro");
+
+    String file3 = new File(factory.create(checkpointId + 1).location()).getName();
+    assertThat(file3).isEqualTo("job123-operator456-00007-2-101-00003.avro");
+  }
+
+  @Test
+  public void testFileNameFormatWithSuffix() {
+    String flinkJobId = "job123";
+    String operatorUniqueId = "operator456";
+    int subTaskId = 7;
+    long attemptNumber = 2;
+    long checkpointId = 100;
+    Map<String, String> props = table.properties();
+
+    ManifestOutputFileFactory factory =
+        new ManifestOutputFileFactory(
+            () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, "suffix");
+
+    String file1 = new File(factory.create(checkpointId).location()).getName();
+    assertThat(file1).isEqualTo("job123-operator456-00007-2-100-00001-suffix.avro");
+
+    String file2 = new File(factory.create(checkpointId).location()).getName();
+    assertThat(file2).isEqualTo("job123-operator456-00007-2-100-00002-suffix.avro");
+  }
+
+  @Test
+  public void testSuffixedFileNamesWithRecreatedFactory() {
+    String flinkJobId = "test-job";
+    String operatorUniqueId = "test-operator";
+    int subTaskId = 0;
+    long attemptNumber = 1;
+    long checkpointId = 1;
+    Map<String, String> props = table.properties();
+
+    ManifestOutputFileFactory factory1 =
+        new ManifestOutputFileFactory(
+            () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, "suffix1");
+    String file1 = new File(factory1.create(checkpointId).location()).getName();
+    assertThat(file1).isEqualTo("test-job-test-operator-00000-1-1-00001-suffix1.avro");
+
+    ManifestOutputFileFactory factory2 =
+        new ManifestOutputFileFactory(
+            () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, "suffix2");
+    String file2 = new File(factory2.create(checkpointId).location()).getName();
+    assertThat(file2).isEqualTo("test-job-test-operator-00000-1-1-00001-suffix2.avro");
+
+    assertThat(file1).isNotEqualTo(file2);
+  }
+}

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java

@@ -68,6 +68,8 @@ class TestDynamicCommitter {
 
   Catalog catalog;
 
+  final int cacheMaximumSize = 10;
+
   private static final DataFile DATA_FILE =
       DataFiles.builder(PartitionSpec.unpartitioned())
           .withPath("/path/to/data-1.parquet")
@@ -155,7 +157,7 @@ void testCommit() throws Exception {
         new WriteTarget(TABLE2, "branch2", 43, 0, true, Sets.newHashSet(1, 2));
 
     DynamicWriteResultAggregator aggregator =
-        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader());
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
     OneInputStreamOperatorTestHarness aggregatorHarness =
         new OneInputStreamOperatorTestHarness(aggregator);
     aggregatorHarness.open();
@@ -281,7 +283,7 @@ void testAlreadyCommitted() throws Exception {
         new WriteTarget(TABLE1, "branch", 42, 0, false, Sets.newHashSet(1, 2));
 
     DynamicWriteResultAggregator aggregator =
-        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader());
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
     OneInputStreamOperatorTestHarness aggregatorHarness =
         new OneInputStreamOperatorTestHarness(aggregator);
     aggregatorHarness.open();
@@ -339,7 +341,7 @@ void testTableBranchAtomicCommitForAppendOnlyData() throws Exception {
     assertThat(table.snapshots()).isEmpty();
 
     DynamicWriteResultAggregator aggregator =
-        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader());
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
     OneInputStreamOperatorTestHarness aggregatorHarness =
         new OneInputStreamOperatorTestHarness(aggregator);
     aggregatorHarness.open();
@@ -454,7 +456,7 @@ void testTableBranchAtomicCommitWithFailures() throws Exception {
     assertThat(table.snapshots()).isEmpty();
 
     DynamicWriteResultAggregator aggregator =
-        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader());
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
     OneInputStreamOperatorTestHarness aggregatorHarness =
         new OneInputStreamOperatorTestHarness(aggregator);
     aggregatorHarness.open();
@@ -613,7 +615,7 @@ void testReplacePartitions() throws Exception {
         new WriteTarget(TABLE1, "branch", 42, 0, false, Sets.newHashSet(1, 2));
 
     DynamicWriteResultAggregator aggregator =
-        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader());
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
     OneInputStreamOperatorTestHarness aggregatorHarness =
         new OneInputStreamOperatorTestHarness(aggregator);
     aggregatorHarness.open();