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: [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]
