(iceberg) branch main updated: Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater (#18419)

[mxm]

This is an automated email from the ASF dual-hosted git repository.

mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 067db056d0 Flink 2.3: Fix DynamicIcebergSink branch creation race in 
TableUpdater (#18419)
067db056d0 is described below

commit 067db056d0ce1005a58a5d1fb15d98499f809df3
Author: Anton Kozmirchuk <kozmirc...@gmail.com>
AuthorDate: Thu Oct 8 14:40:09 2026 +0100

    Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater 
(#18419)
---
 .../iceberg/flink/sink/dynamic/TableUpdater.java   |  7 +++++--
 .../flink/sink/dynamic/TestTableUpdater.java       | 22 ++++++++++++++++++++++
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
 
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 26eb34db3c..25efa744a7 100644
--- 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++ 
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -107,8 +107,11 @@ class TableUpdater {
       try {
         table.manageSnapshots().createBranch(branch).commit();
         LOG.info("Branch {} for {} created", branch, identifier);
-      } catch (CommitFailedException e) {
-        table.refresh();
+      } catch (CommitFailedException | IllegalArgumentException e) {
+        if (!table.refs().containsKey(branch)) {
+          table.refresh();
+        }
+
         if (table.refs().containsKey(branch)) {
           LOG.debug("Branch {} concurrently created for {}.", branch, 
identifier);
         } else {
diff --git 
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
 
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 48c0ffda64..367250cf27 100644
--- 
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++ 
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
@@ -117,6 +117,28 @@ public class TestTableUpdater extends 
TestFlinkIcebergSinkBase {
     assertThat(result.f1).isEqualTo(PartitionSpec.unpartitioned());
   }
 
+  @Test
+  void testBranchConcurrentlyCreatedByAnotherWriter() {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    TableIdentifier tableIdentifier = TableIdentifier.parse("myTable");
+    catalog.createTable(tableIdentifier, SCHEMA);
+
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, 
CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, 
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // This writer's cache while the branch genuinely does not exist yet.
+    assertThat(cache.branch(tableIdentifier, "myBranch")).isNull();
+
+    // Simulate another writer subtask committing the branch
+    
catalog.loadTable(tableIdentifier).manageSnapshots().createBranch("myBranch").commit();
+
+    // Must not throw: a branch concurrently created by another writer must be 
treated success
+    tableUpdater.update(
+        tableIdentifier, "myBranch", SCHEMA, PartitionSpec.unpartitioned(), 
TableCreator.DEFAULT);
+    assertThat(cache.branch(tableIdentifier, "myBranch")).isNotNull();
+  }
+
   @Test
   void testBranchCreationAndCaching() {
     Catalog catalog = CATALOG_EXTENSION.catalog();