You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "kozmirchuk (via GitHub)" <gi...@apache.org> on 2026/10/08 10:19:06 UTC
[PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
kozmirchuk opened a new pull request, #18419:
URL: https://github.com/apache/iceberg/pull/18419
When several writer subtasks create the same branch concurrently, `TableUpdater#findOrCreateBranch` only treated `CommitFailedException` as "branch already created".
`createBranch(...).commit()` can also fail with `IllegalArgumentException` ("Ref ... already exists") when its metadata refresh already sees the other writer's branch, before any commit is attempted.
Test plan: `testBranchConcurrentlyCreatedByAnotherWriter` caches the branch as missing, creates it through another table instance, then calls `update`. It fails without the fix and passes with it.
---
**AI Disclosure**
- Model: Claude Opus 5.5
- Platform/Tool: Claude Code
- Human Oversight: Reviewed at every step
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "kozmirchuk (via GitHub)" <gi...@apache.org>.
kozmirchuk commented on PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#issuecomment-6061219766
@mxm Sure! I'm on it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4219235435
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,8 +107,11 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
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();
+ }
+
Review Comment:
Could we move this check after the `loadTable`? I think that way we should be able to remove the IllegalArgumentException entirely because we do not refresh the table locally after the loadTable call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "kozmirchuk (via GitHub)" <gi...@apache.org>.
kozmirchuk commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4219432048
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,8 +107,11 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
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();
+ }
+
Review Comment:
`table.manageSnapshot()` does a refresh so the following scenario is possible:
Lets say w1 and w2 are writer subtasks.
1. w1 and w2 both have cold caches, so both enter findOrCreateBranch.
2. w2 calls loadTable, which returns metadata with no main ref.
3. w1 calls `manageSnapshots().createBranch("main").commit()`, which succeeds.
4. w2 calls `table.manageSnapshots()`. The `SnapshotManager` constructor does `ops.refresh()`, which re-reads the catalog and picks up w1's main.
5. w2's `createBranch("main")` finds the ref and throws `IllegalArgumentException: Ref main already exists`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4217865171
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,7 +107,7 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
try {
table.manageSnapshots().createBranch(branch).commit();
LOG.info("Branch {} for {} created", branch, identifier);
- } catch (CommitFailedException e) {
+ } catch (CommitFailedException | IllegalArgumentException e) {
table.refresh();
if (table.refs().containsKey(branch)) {
Review Comment:
NIT: It could help to check the table refs before refreshing the table. I think the `IllegalArgumentException` can only happen when the table already has the branch information from the `loadTable` call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "kozmirchuk (via GitHub)" <gi...@apache.org>.
kozmirchuk commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4218973985
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,7 +107,7 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
try {
table.manageSnapshots().createBranch(branch).commit();
LOG.info("Branch {} for {} created", branch, identifier);
- } catch (CommitFailedException e) {
+ } catch (CommitFailedException | IllegalArgumentException e) {
table.refresh();
if (table.refs().containsKey(branch)) {
Review Comment:
This makes sense. Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4217872876
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,7 +107,7 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
try {
table.manageSnapshots().createBranch(branch).commit();
LOG.info("Branch {} for {} created", branch, identifier);
- } catch (CommitFailedException e) {
+ } catch (CommitFailedException | IllegalArgumentException e) {
table.refresh();
if (table.refs().containsKey(branch)) {
Review Comment:
This is probably more than a NIT because we would save a `table.refresh()` call in that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4219054052
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -117,6 +117,27 @@ void testTableAlreadyExists() {
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.
+ cache.branch(tableIdentifier, "myBranch");
Review Comment:
Should this be:
```suggestion
assertThat(cache.branch(tableIdentifier, "myBranch")).isNull();
```
?
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -117,6 +117,27 @@ void testTableAlreadyExists() {
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.
+ cache.branch(tableIdentifier, "myBranch");
+
+ // 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);
Review Comment:
Final verification:
```suggestion
tableIdentifier, "myBranch", SCHEMA, PartitionSpec.unpartitioned(), TableCreator.DEFAULT);
assertThat(cache.branch(tableIdentifier, "myBranch")).isNotNull();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#issuecomment-6061164567
Thanks @kozmirchuk! Could you please open a backport PR for the other Flink versions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm merged PR #18419:
URL: https://github.com/apache/iceberg/pull/18419
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "kozmirchuk (via GitHub)" <gi...@apache.org>.
kozmirchuk commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4219129737
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -117,6 +117,27 @@ void testTableAlreadyExists() {
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.
+ cache.branch(tableIdentifier, "myBranch");
Review Comment:
Done
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -117,6 +117,27 @@ void testTableAlreadyExists() {
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.
+ cache.branch(tableIdentifier, "myBranch");
+
+ // 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);
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink 2.3: Fix DynamicIcebergSink branch creation race in TableUpdater [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18419:
URL: https://github.com/apache/iceberg/pull/18419#discussion_r4219524646
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -107,8 +107,11 @@ private void findOrCreateBranch(TableIdentifier identifier, String branch) {
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();
+ }
+
Review Comment:
Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org