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 0222e8bb4d Flink: Backport: Fix DynamicIcebergSink branch creation
race in TableUpdater (#18422)
0222e8bb4d is described below
commit 0222e8bb4d3b338c4366426356fe86a60a09c6de
Author: Anton Kozmirchuk <[email protected]>
AuthorDate: Thu Oct 8 16:19:48 2026 +0100
Flink: Backport: Fix DynamicIcebergSink branch creation race in
TableUpdater (#18422)
---
.../iceberg/flink/sink/dynamic/TableUpdater.java | 7 +++++--
.../flink/sink/dynamic/TestTableUpdater.java | 22 ++++++++++++++++++++++
.../iceberg/flink/sink/dynamic/TableUpdater.java | 7 +++++--
.../flink/sink/dynamic/TestTableUpdater.java | 22 ++++++++++++++++++++++
4 files changed, 54 insertions(+), 4 deletions(-)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 26eb34db3c..25efa744a7 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v1.20/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/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 48c0ffda64..367250cf27 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++
b/flink/v1.20/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();
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 26eb34db3c..25efa744a7 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v2.2/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.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 48c0ffda64..367250cf27 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++
b/flink/v2.2/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();
