From 155b9c051ee6ae1794bef6caceee26ea2728d330 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 16 Oct 2025 13:35:07 +0200 Subject: [PATCH 1/5] Flink: Prevent recreation of ManifestOutputFileFactory during flushing DynamicWriteResultAggregator uses the ManifestOutputFileFactory class to write a temporary manifest. For the Dynamic Sink we want to support writing to a vast amount of tables, even during a single checkpoint. So we avoid storing all factories and use a cache with an eviction policy. The problem is that if the factory for a given table is evicted during a checkpoint flush while there could still be writes for that factory being processed. In that case the same output directory will be generated again which leads to overwriting already written manifests files. We must avoid recreating the output file factory during checkpoint flushing. It is fine to drop the factories due to cache eviction afterwards, as the output paths for factories are scoped by checkpoint id. --- .../sink/dynamic/DynamicIcebergSink.java | 2 +- .../dynamic/DynamicWriteResultAggregator.java | 32 ++++--- .../iceberg/flink/sink/dynamic/LRUCache.java | 19 +++- .../sink/dynamic/TestDynamicCommitter.java | 12 ++- .../TestDynamicWriteResultAggregator.java | 92 ++++++++++++++++++- .../flink/sink/dynamic/TestLRUCache.java | 63 +++++++++++++ 6 files changed, 199 insertions(+), 21 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java index 2715a01608d6..db48be7977f0 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java @@ -168,7 +168,7 @@ public DataStream> addPreCommitTopology( .transform( prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"), typeInformation, - new DynamicWriteResultAggregator(catalogLoader)) + new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize)) .uid(prefixIfNotNull(uidPrefix, sinkId + "-pre-commit-topology")); } diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java index c833527ecc55..9394bbfcdc3e 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java @@ -18,10 +18,7 @@ */ 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 org.apache.flink.annotation.VisibleForTesting; @@ -61,23 +58,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> results; - private transient Cache> specs; - private transient Cache outputFileFactories; + private transient Map> specs; + private transient LRUCache 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 +92,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(); } @@ -118,6 +114,14 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws IOException { this.lastCheckpointId = checkpointId; + // Pause eviction on the cache for ManifestOutputFileFactory. + // This will materialize a list of all output file factories required to write delta manifests. + // Some of the output file factories may already be cached, and we will reuse those. We must + // absolutely avoid re-creating any output file factories _during_ writing manifests, otherwise + // cache eviction may reset the manifest file names for multiple WriteResults for a given table + // which will overwrite already written manifest files! + outputFileFactories.haltEviction(); + Collection> committables = Sets.newHashSetWithExpectedSize(results.size()); int count = 0; @@ -145,6 +149,8 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws IOException { new CommittableWithLineage<>(c.getCommittable(), checkpointId, subTaskId)))); LOG.info("Emitted {} commit message to downstream committer operator", count); results.clear(); + // Continue eviction and clean up + outputFileFactories.continueEviction(); } /** @@ -186,7 +192,7 @@ public void processElement(StreamRecord> } private ManifestOutputFileFactory outputFileFactory(String tableName) { - return outputFileFactories.get( + return outputFileFactories.computeIfAbsent( tableName, unused -> { Table table = catalog.loadTable(TableIdentifier.parse(tableName)); @@ -197,7 +203,7 @@ private ManifestOutputFileFactory outputFileFactory(String tableName) { } private PartitionSpec spec(String tableName, int specId) { - Map knownSpecs = specs.getIfPresent(tableName); + Map knownSpecs = specs.get(tableName); if (knownSpecs != null) { PartitionSpec spec = knownSpecs.get(specId); if (spec != null) { diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java index be2866dc4e19..4f5b1055aa4e 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.flink.sink.dynamic; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; @@ -42,6 +43,8 @@ class LRUCache extends LinkedHashMap { private final int maximumSize; private final Consumer> evictionCallback; + private boolean evictEldest = true; + LRUCache(int maximumSize) { this(maximumSize, ignored -> {}); } @@ -54,11 +57,25 @@ class LRUCache extends LinkedHashMap { @Override protected boolean removeEldestEntry(Map.Entry eldest) { - boolean remove = size() > maximumSize; + boolean remove = size() > maximumSize && evictEldest; if (remove) { evictionCallback.accept(eldest); } return remove; } + + void haltEviction() { + evictEldest = false; + } + + void continueEviction() { + evictEldest = true; + Iterator> iterator = this.entrySet().iterator(); + while (size() > maximumSize) { + Map.Entry entry = iterator.next(); + evictionCallback.accept(entry); + iterator.remove(); + } + } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java index f5387aee882a..7894428a781f 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java +++ b/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(); diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java index 713c67da170a..d3f4385d972b 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java @@ -20,15 +20,26 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.io.IOException; +import java.util.List; +import java.util.Set; +import org.apache.flink.core.io.SimpleVersionedSerialization; import org.apache.flink.streaming.api.connector.sink2.CommittableMessage; +import org.apache.flink.streaming.api.connector.sink2.CommittableSummary; import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness; import org.apache.hadoop.util.Sets; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.flink.HadoopCatalogExtension; +import org.apache.iceberg.flink.sink.DeltaManifests; +import org.apache.iceberg.flink.sink.DeltaManifestsSerializer; import org.apache.iceberg.io.WriteResult; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -37,13 +48,22 @@ class TestDynamicWriteResultAggregator { @RegisterExtension static final HadoopCatalogExtension CATALOG_EXTENSION = new HadoopCatalogExtension("db", "table"); + final int cacheMaximumSize = 1; + + private static final DataFile DATA_FILE = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath("/path/to/data-1.parquet") + .withFileSizeInBytes(100) + .withRecordCount(1) + .build(); + @Test void testAggregator() throws Exception { CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), new Schema()); CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table2"), new Schema()); DynamicWriteResultAggregator aggregator = - new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader()); + new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize); try (OneInputStreamOperatorTestHarness< CommittableMessage, CommittableMessage> testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) { @@ -79,4 +99,74 @@ void testAggregator() throws Exception { assertThat(testHarness.getRecordOutput()).hasSize(4); } } + + @Test + void testPreventOutputFileFactoryCacheEvictionDuringFlush() throws Exception { + CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), new Schema()); + + // Disable caching of ManifestOutputFileFactory. + final int zeroCacheSize = 0; + DynamicWriteResultAggregator aggregator = + new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), zeroCacheSize); + try (OneInputStreamOperatorTestHarness< + CommittableMessage, CommittableMessage> + testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) { + testHarness.open(); + + WriteTarget writeTarget1 = + new WriteTarget("table", "branch", 42, 0, false, Sets.newHashSet()); + DynamicWriteResult dynamicWriteResult1 = + new DynamicWriteResult( + writeTarget1, WriteResult.builder().addDataFiles(DATA_FILE).build()); + + // Different WriteTarget + WriteTarget writeTarget2 = + new WriteTarget("table", "branch2", 23, 0, true, Sets.newHashSet()); + DynamicWriteResult dynamicWriteResult2 = + new DynamicWriteResult( + writeTarget2, WriteResult.builder().addDataFiles(DATA_FILE).build()); + + CommittableWithLineage committable1 = + new CommittableWithLineage<>(dynamicWriteResult1, 0, 0); + testHarness.processElement(new StreamRecord<>(committable1)); + + CommittableWithLineage committable2 = + new CommittableWithLineage<>(dynamicWriteResult2, 0, 0); + testHarness.processElement(new StreamRecord<>(committable2)); + + assertThat(testHarness.getOutput()).isEmpty(); + + testHarness.prepareSnapshotPreBarrier(1L); + List>> output = + Lists.newArrayList(testHarness.getRecordOutput().iterator()); + + assertThat(testHarness.getOutput()).hasSize(3); + assertThat(output.get(0).getValue()).isInstanceOf(CommittableSummary.class); + assertThat(output.get(1).getValue()).isInstanceOf(CommittableWithLineage.class); + assertThat(output.get(2).getValue()).isInstanceOf(CommittableWithLineage.class); + + // There should be two unique file paths, despite the cache being disabled. + Set manifestPaths = getManifestPaths(output); + assertThat(manifestPaths).hasSize(2); + } + } + + private static Set getManifestPaths( + List>> messages) throws IOException { + Set manifestPaths = Sets.newHashSet(); + + for (StreamRecord> record : messages) { + CommittableMessage message = record.getValue(); + if (message instanceof CommittableWithLineage) { + DeltaManifests deltaManifests = + SimpleVersionedSerialization.readVersionAndDeSerialize( + DeltaManifestsSerializer.INSTANCE, + (((CommittableWithLineage) message).getCommittable()) + .manifest()); + deltaManifests.manifests().forEach(manifest -> manifestPaths.add(manifest.path())); + } + } + + return manifestPaths; + } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java index 679d3de978a3..7d6d6fd0ac1d 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java @@ -77,6 +77,69 @@ void testEvictionCallback() { assertThat(callback.evictedEntries).containsExactly(Map.entry(2, 2)); } + @Test + void testHaltEviction() { + int maxSize = 2; + LRUCache cache = new LRUCache<>(maxSize, NO_OP_CALLBACK); + + cache.put(1, 1); + cache.put(2, 2); + + cache.haltEviction(); + + cache.put(3, 3); + cache.put(4, 4); + + assertThat(cache) + .hasSize(4) + .containsEntry(1, 1) + .containsEntry(2, 2) + .containsEntry(3, 3) + .containsEntry(4, 4); + } + + @Test + void testContinueEviction() { + int maxSize = 2; + LRUCache cache = new LRUCache<>(maxSize, NO_OP_CALLBACK); + + cache.put(1, 1); + cache.put(2, 2); + + cache.haltEviction(); + + cache.put(3, 3); + cache.put(4, 4); + + assertThat(cache).hasSize(4); + + cache.continueEviction(); + + assertThat(cache).hasSize(2).containsEntry(3, 3).containsEntry(4, 4); + } + + @Test + void testContinueEvictionWithCallback() { + int maxSize = 2; + TestEvictionCallback callback = new TestEvictionCallback(); + LRUCache cache = new LRUCache<>(maxSize, callback); + + cache.put(1, 1); + cache.put(2, 2); + + cache.haltEviction(); + + cache.put(3, 3); + cache.put(4, 4); + + assertThat(callback.evictedEntries).isEmpty(); + + cache.continueEviction(); + + assertThat(cache).hasSize(2); + assertThat(callback.evictedEntries).containsExactly(Map.entry(1, 1), Map.entry(2, 2)); + } + private static class TestEvictionCallback implements Consumer> { private final List> evictedEntries = Lists.newArrayList(); From d0ceaf73b413a37930d28568db0b5b0c7a1702c9 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Fri, 17 Oct 2025 12:50:01 +0200 Subject: [PATCH 2/5] Change implementation to use a UUID for file names to allow factory re-creation during flushing --- flink/v2.1/build.gradle | 3 - .../flink/sink/ManifestOutputFileFactory.java | 9 +- .../dynamic/DynamicWriteResultAggregator.java | 10 --- .../iceberg/flink/sink/dynamic/LRUCache.java | 19 +--- .../flink/sink/TestIcebergCommitter.java | 4 +- .../flink/sink/TestIcebergFilesCommitter.java | 8 +- .../sink/TestManifestOutputFileFactory.java | 87 +++++++++++++++++++ .../flink/sink/dynamic/TestLRUCache.java | 63 -------------- 8 files changed, 99 insertions(+), 104 deletions(-) create mode 100644 flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java diff --git a/flink/v2.1/build.gradle b/flink/v2.1/build.gradle index 224f0a555e5f..91081bdc2e42 100644 --- a/flink/v2.1/build.gradle +++ b/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 diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java index 6ba87bea30c2..b739f3c6fb81 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java @@ -20,7 +20,7 @@ import java.util.Locale; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.UUID; import java.util.function.Supplier; import org.apache.flink.annotation.Internal; import org.apache.iceberg.FileFormat; @@ -42,7 +42,6 @@ public class ManifestOutputFileFactory { private final String operatorUniqueId; private final int subTaskId; private final long attemptNumber; - private final AtomicInteger fileCount = new AtomicInteger(0); ManifestOutputFileFactory( Supplier tableSupplier, @@ -63,13 +62,15 @@ private String generatePath(long checkpointId) { return FileFormat.AVRO.addExtension( String.format( Locale.ROOT, - "%s-%s-%05d-%d-%d-%05d", + "%s-%s-%05d-%d-%d-%s", flinkJobId, operatorUniqueId, subTaskId, attemptNumber, checkpointId, - fileCount.incrementAndGet())); + // Make sure to append a random identifier in case this factory were to get re-created + // during a checkpoint, e.g. due to cache eviction. + UUID.randomUUID())); } public OutputFile create(long checkpointId) { diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java index 9394bbfcdc3e..507ee683dc06 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java @@ -114,14 +114,6 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws IOException { this.lastCheckpointId = checkpointId; - // Pause eviction on the cache for ManifestOutputFileFactory. - // This will materialize a list of all output file factories required to write delta manifests. - // Some of the output file factories may already be cached, and we will reuse those. We must - // absolutely avoid re-creating any output file factories _during_ writing manifests, otherwise - // cache eviction may reset the manifest file names for multiple WriteResults for a given table - // which will overwrite already written manifest files! - outputFileFactories.haltEviction(); - Collection> committables = Sets.newHashSetWithExpectedSize(results.size()); int count = 0; @@ -149,8 +141,6 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws IOException { new CommittableWithLineage<>(c.getCommittable(), checkpointId, subTaskId)))); LOG.info("Emitted {} commit message to downstream committer operator", count); results.clear(); - // Continue eviction and clean up - outputFileFactories.continueEviction(); } /** diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java index 4f5b1055aa4e..be2866dc4e19 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/LRUCache.java @@ -18,7 +18,6 @@ */ package org.apache.iceberg.flink.sink.dynamic; -import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; @@ -43,8 +42,6 @@ class LRUCache extends LinkedHashMap { private final int maximumSize; private final Consumer> evictionCallback; - private boolean evictEldest = true; - LRUCache(int maximumSize) { this(maximumSize, ignored -> {}); } @@ -57,25 +54,11 @@ class LRUCache extends LinkedHashMap { @Override protected boolean removeEldestEntry(Map.Entry eldest) { - boolean remove = size() > maximumSize && evictEldest; + boolean remove = size() > maximumSize; if (remove) { evictionCallback.accept(eldest); } return remove; } - - void haltEviction() { - evictEldest = false; - } - - void continueEviction() { - evictEldest = true; - Iterator> iterator = this.entrySet().iterator(); - while (size() > maximumSize) { - Map.Entry entry = iterator.next(); - evictionCallback.accept(entry); - iterator.remove(); - } - } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java index abac7e16d9f0..184726bca004 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java @@ -1016,8 +1016,8 @@ public void testFlinkManifests() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .isEqualTo( - String.format("%s-%s-%05d-%d-%d-%05d.avro", jobID, operatorId, 0, 0, checkpoint, 1)); + .startsWith(String.format("%s-%s-%05d-%d-%d-", jobID, operatorId, 0, 0, checkpoint)) + .endsWith(".avro"); // // 2. Read the data files from manifests and assert. List dataFiles = diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java index 65fb9b8f69b4..f9162df9c89e 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java @@ -757,8 +757,8 @@ public void testFlinkManifests() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .isEqualTo( - String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1)); + .startsWith(String.format("%s-%s-%05d-%d-%d-", jobId, operatorId, 0, 0, checkpoint)) + .endsWith(".avro"); // 2. Read the data files from manifests and assert. List dataFiles = @@ -807,8 +807,8 @@ public void testDeleteFiles() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .isEqualTo( - String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1)); + .startsWith(String.format("%s-%s-%05d-%d-%d-", jobId, operatorId, 0, 0, checkpoint)) + .endsWith(".avro"); // 2. Read the data files from manifests and assert. List dataFiles = diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java new file mode 100644 index 000000000000..2d1bd6a7d4b8 --- /dev/null +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java @@ -0,0 +1,87 @@ +/* + * 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.apache.iceberg.io.OutputFile; +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 props = table.properties(); + + ManifestOutputFileFactory factory = + new ManifestOutputFileFactory( + () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber); + OutputFile file = factory.create(checkpointId); + + String filename = new File(file.location()).getName(); + assertThat(filename).startsWith("job123-operator456-00007-2-100-"); + assertThat(filename).endsWith(".avro"); + } + + @Test + public void testUniqueFileNamesWithRecreatedFactory() { + String flinkJobId = "test-job"; + String operatorUniqueId = "test-operator"; + int subTaskId = 0; + long attemptNumber = 1; + long checkpointId = 1; + Map props = table.properties(); + + ManifestOutputFileFactory factory1 = + new ManifestOutputFileFactory( + () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber); + OutputFile file1 = factory1.create(checkpointId); + + ManifestOutputFileFactory factory2 = + new ManifestOutputFileFactory( + () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber); + OutputFile file2 = factory2.create(checkpointId); + + assertThat(file1.location()).isNotEqualTo(file2.location()); + } +} diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java index 7d6d6fd0ac1d..679d3de978a3 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestLRUCache.java @@ -77,69 +77,6 @@ void testEvictionCallback() { assertThat(callback.evictedEntries).containsExactly(Map.entry(2, 2)); } - @Test - void testHaltEviction() { - int maxSize = 2; - LRUCache cache = new LRUCache<>(maxSize, NO_OP_CALLBACK); - - cache.put(1, 1); - cache.put(2, 2); - - cache.haltEviction(); - - cache.put(3, 3); - cache.put(4, 4); - - assertThat(cache) - .hasSize(4) - .containsEntry(1, 1) - .containsEntry(2, 2) - .containsEntry(3, 3) - .containsEntry(4, 4); - } - - @Test - void testContinueEviction() { - int maxSize = 2; - LRUCache cache = new LRUCache<>(maxSize, NO_OP_CALLBACK); - - cache.put(1, 1); - cache.put(2, 2); - - cache.haltEviction(); - - cache.put(3, 3); - cache.put(4, 4); - - assertThat(cache).hasSize(4); - - cache.continueEviction(); - - assertThat(cache).hasSize(2).containsEntry(3, 3).containsEntry(4, 4); - } - - @Test - void testContinueEvictionWithCallback() { - int maxSize = 2; - TestEvictionCallback callback = new TestEvictionCallback(); - LRUCache cache = new LRUCache<>(maxSize, callback); - - cache.put(1, 1); - cache.put(2, 2); - - cache.haltEviction(); - - cache.put(3, 3); - cache.put(4, 4); - - assertThat(callback.evictedEntries).isEmpty(); - - cache.continueEviction(); - - assertThat(cache).hasSize(2); - assertThat(callback.evictedEntries).containsExactly(Map.entry(1, 1), Map.entry(2, 2)); - } - private static class TestEvictionCallback implements Consumer> { private final List> evictedEntries = Lists.newArrayList(); From 586f90b79fadc45884cfdaf71c88b785ead5e35d Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Fri, 17 Oct 2025 13:55:41 +0200 Subject: [PATCH 3/5] fixup! Use super type --- .../flink/sink/dynamic/DynamicWriteResultAggregator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java index 507ee683dc06..3585d08f3f5e 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java @@ -66,7 +66,7 @@ class DynamicWriteResultAggregator private transient Map> results; private transient Map> specs; - private transient LRUCache outputFileFactories; + private transient Map outputFileFactories; private transient String flinkJobId; private transient String operatorId; private transient int subTaskId; From 04e51ba6d2a44b4a3bf71241b1777a54202823a4 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 20 Oct 2025 11:34:39 +0200 Subject: [PATCH 4/5] Preserve old file names for non-dynamic sink --- .../iceberg/flink/sink/FlinkManifestUtil.java | 14 +++++- .../flink/sink/ManifestOutputFileFactory.java | 16 +++--- .../dynamic/DynamicWriteResultAggregator.java | 12 ++++- .../iceberg/flink/sink/TestFlinkManifest.java | 2 +- .../flink/sink/TestIcebergCommitter.java | 4 +- .../flink/sink/TestIcebergFilesCommitter.java | 8 +-- .../sink/TestManifestOutputFileFactory.java | 49 ++++++++++++++----- 7 files changed, 78 insertions(+), 27 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java index 13affd8484aa..0eeedf265914 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java +++ b/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
tableSupplier, + Map tableProps, + String flinkJobId, + String operatorUniqueId, + int subTaskId, + long attemptNumber, + String suffix) { + return new ManifestOutputFileFactory( + tableSupplier, tableProps, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, suffix); } /** diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java index b739f3c6fb81..30c0e11a2554 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java @@ -20,8 +20,9 @@ import java.util.Locale; import java.util.Map; -import java.util.UUID; +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; @@ -42,6 +43,8 @@ public class ManifestOutputFileFactory { private final String operatorUniqueId; private final int subTaskId; private final long attemptNumber; + private final AtomicInteger fileCount = new AtomicInteger(0); + @Nullable private final String suffix; ManifestOutputFileFactory( Supplier
tableSupplier, @@ -49,28 +52,29 @@ public class ManifestOutputFileFactory { 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-%s", + "%s-%s-%05d-%d-%d-%05d%s", flinkJobId, operatorUniqueId, subTaskId, attemptNumber, checkpointId, - // Make sure to append a random identifier in case this factory were to get re-created - // during a checkpoint, e.g. due to cache eviction. - UUID.randomUUID())); + fileCount.incrementAndGet(), + suffix != null ? "-" + suffix : "")); } public OutputFile create(long checkpointId) { diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java index 3585d08f3f5e..2c35ee0fdb21 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java @@ -21,6 +21,7 @@ import java.io.IOException; 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; @@ -187,8 +188,17 @@ private ManifestOutputFileFactory outputFileFactory(String 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 suffix = UUID.randomUUID().toString(); return FlinkManifestUtil.createOutputFileFactory( - () -> table, table.properties(), flinkJobId, operatorId, subTaskId, attemptId); + () -> table, + table.properties(), + flinkJobId, + operatorId, + subTaskId, + attemptId, + suffix); }); } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkManifest.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkManifest.java index c21c3d5cc21b..2c5fd4b0509d 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkManifest.java +++ b/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 dataFiles = generateDataFiles(5); DeltaManifests deltaManifests = diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java index 184726bca004..abac7e16d9f0 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java @@ -1016,8 +1016,8 @@ public void testFlinkManifests() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .startsWith(String.format("%s-%s-%05d-%d-%d-", jobID, operatorId, 0, 0, checkpoint)) - .endsWith(".avro"); + .isEqualTo( + String.format("%s-%s-%05d-%d-%d-%05d.avro", jobID, operatorId, 0, 0, checkpoint, 1)); // // 2. Read the data files from manifests and assert. List dataFiles = diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java index f9162df9c89e..65fb9b8f69b4 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java @@ -757,8 +757,8 @@ public void testFlinkManifests() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .startsWith(String.format("%s-%s-%05d-%d-%d-", jobId, operatorId, 0, 0, checkpoint)) - .endsWith(".avro"); + .isEqualTo( + String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1)); // 2. Read the data files from manifests and assert. List dataFiles = @@ -807,8 +807,8 @@ public void testDeleteFiles() throws Exception { Path manifestPath = manifestPaths.get(0); assertThat(manifestPath.getFileName()) .asString() - .startsWith(String.format("%s-%s-%05d-%d-%d-", jobId, operatorId, 0, 0, checkpoint)) - .endsWith(".avro"); + .isEqualTo( + String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1)); // 2. Read the data files from manifests and assert. List dataFiles = diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java index 2d1bd6a7d4b8..654248fcabd9 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestManifestOutputFileFactory.java @@ -27,7 +27,6 @@ import org.apache.iceberg.Table; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.flink.HadoopCatalogExtension; -import org.apache.iceberg.io.OutputFile; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -55,16 +54,40 @@ public void testFileNameFormat() { ManifestOutputFileFactory factory = new ManifestOutputFileFactory( - () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber); - OutputFile file = factory.create(checkpointId); + () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber, null); - String filename = new File(file.location()).getName(); - assertThat(filename).startsWith("job123-operator456-00007-2-100-"); - assertThat(filename).endsWith(".avro"); + 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 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 testUniqueFileNamesWithRecreatedFactory() { + public void testSuffixedFileNamesWithRecreatedFactory() { String flinkJobId = "test-job"; String operatorUniqueId = "test-operator"; int subTaskId = 0; @@ -74,14 +97,16 @@ public void testUniqueFileNamesWithRecreatedFactory() { ManifestOutputFileFactory factory1 = new ManifestOutputFileFactory( - () -> table, props, flinkJobId, operatorUniqueId, subTaskId, attemptNumber); - OutputFile file1 = factory1.create(checkpointId); + () -> 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); - OutputFile file2 = factory2.create(checkpointId); + () -> 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.location()).isNotEqualTo(file2.location()); + assertThat(file1).isNotEqualTo(file2); } } From 3f01fcb7626d872342e80f3d081781748da83211 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 20 Oct 2025 13:59:45 +0200 Subject: [PATCH 5/5] Verify that DynamicWriter does not suffer from the same issue --- .../dynamic/DynamicWriteResultAggregator.java | 4 +-- .../flink/sink/dynamic/DynamicWriter.java | 5 +++ .../flink/sink/dynamic/TestDynamicWriter.java | 32 +++++++++++++++++++ 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java index 2c35ee0fdb21..b3262957ad37 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java @@ -190,7 +190,7 @@ private ManifestOutputFileFactory outputFileFactory(String 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 suffix = UUID.randomUUID().toString(); + String fileSuffix = UUID.randomUUID().toString(); return FlinkManifestUtil.createOutputFileFactory( () -> table, table.properties(), @@ -198,7 +198,7 @@ private ManifestOutputFileFactory outputFileFactory(String tableName) { operatorId, subTaskId, attemptId, - suffix); + fileSuffix); }); } diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java index ae24efafa6af..e99e6e72da21 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java @@ -210,4 +210,9 @@ private static Set getEqualityFields(Table table, Set equality DynamicWriterMetrics getMetrics() { return metrics; } + + @VisibleForTesting + Map getTaskWriterFactories() { + return taskWriterFactories; + } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java index 91a3a5d5a7c6..91a6883932c9 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java @@ -201,6 +201,38 @@ void testDynamicWriterUpsertNoEqualityFields() { "Equality field columns shouldn't be empty when configuring to use UPSERT data."); } + @Test + void testUniqueFileSuffixOnFactoryRecreation() throws Exception { + Catalog catalog = CATALOG_EXTENSION.catalog(); + Table table1 = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA); + + DynamicWriter dynamicWriter = createDynamicWriter(catalog); + DynamicRecordInternal record1 = getDynamicRecordInternal(table1); + + dynamicWriter.write(record1, null); + dynamicWriter.prepareCommit(); + + File dataDir1 = new File(URI.create(table1.location()).getPath(), "data"); + File[] files = dataDir1.listFiles((dir, name) -> !name.startsWith(".")); + assertThat(files).isNotNull().hasSize(1); + File firstFile = files[0]; + + // Clear cache which must create new unique files names for the output files + dynamicWriter.getTaskWriterFactories().clear(); + + dynamicWriter.write(record1, null); + dynamicWriter.prepareCommit(); + + files = + dataDir1.listFiles( + (dir, name) -> !name.startsWith(".") && !name.equals(firstFile.getName())); + assertThat(files).isNotNull().hasSize(1); + File secondFile = files[0]; + + // File names must be different + assertThat(firstFile.getName()).isNotEqualTo(secondFile.getName()); + } + private static @Nonnull DynamicWriter createDynamicWriter( Catalog catalog, Map properties) { DynamicWriter dynamicWriter =