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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
