This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 92e39e101b [iceberg] Make REST catalog namespace creation idempotent
under concurrent commits (#8471)
92e39e101b is described below
commit 92e39e101bbcfd8bb287f345a07be1aeac2b3022
Author: Andreas Bube <[email protected]>
AuthorDate: Mon Jul 6 14:47:13 2026 +0200
[iceberg] Make REST catalog namespace creation idempotent under concurrent
commits (#8471)
When two Paimon sinks commit to different tables in the same Iceberg
REST catalog database, and that database (namespace) does not yet exist,
they can race on namespace creation.
`IcebergRestMetadataCommitter.commitMetadataImpl()` uses a
check-then-act pattern across two separate REST round-trips:
```java
// create database if not exist
if (!databaseExists()) {
createDatabase();
}
```
Both commits can observe the namespace as missing and both call
`createNamespace()`. The loser fails — per the Iceberg REST spec with a
409 `AlreadyExistsException` (as `RESTServerExtension` returns), or with
a 500 `ServiceFailureException` on the non-spec-compliant
`apache/iceberg-rest-fixture` docker image. This causes a spurious
commit failure and sink restart. It recovers after the restart (the
namespace exists by then), but produces noise on the first commit of a
fresh database.
This makes namespace creation idempotent: `createDatabase()` now catches
`AlreadyExistsException` and proceeds, since the required namespace now
exists. (A spec-violating 500 from a non-compliant server is a separate,
server-side problem and is intentionally not swallowed.)
---
.../iceberg/IcebergRestMetadataCommitter.java | 14 +++++++++++--
.../iceberg/IcebergRestMetadataCommitterTest.java | 24 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 2 deletions(-)
diff --git
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
index 820fb2ff94..fbc352e0c5 100644
---
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
+++
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.iceberg;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
@@ -42,6 +43,7 @@ import org.apache.iceberg.TableMetadataParser;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
@@ -287,8 +289,16 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
return restCatalog.tableExists(icebergTableIdentifier);
}
- private void createDatabase() {
- restCatalog.createNamespace(Namespace.of(icebergDatabaseName));
+ @VisibleForTesting
+ void createDatabase() {
+ try {
+ restCatalog.createNamespace(Namespace.of(icebergDatabaseName));
+ } catch (AlreadyExistsException e) {
+ // Benign check-then-act race: another commit created the
namespace concurrently.
+ LOG.info(
+ "Namespace {} already exists, created concurrently by
another commit.",
+ icebergDatabaseName);
+ }
}
private Table createTable(TableMetadata newMetadata) {
diff --git
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
index a3ae135686..0fbf130a73 100644
---
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
+++
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
@@ -46,6 +46,7 @@ import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.IcebergGenerics;
import org.apache.iceberg.data.Record;
@@ -952,6 +953,29 @@ public class IcebergRestMetadataCommitterTest {
commit.close();
}
+ @Test
+ public void testCreateDatabaseIsIdempotentUnderRace() throws Exception {
+ // Two commits targeting the same namespace can both observe it as
missing and both
+ // call createNamespace() (the check-then-act race in
commitMetadataImpl). The loser
+ // must not fail. Here the second createDatabase() hits an
already-existing namespace.
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ Collections.emptyMap());
+
+ IcebergRestMetadataCommitter committer = new
IcebergRestMetadataCommitter(table);
+ committer.createDatabase();
+ committer.createDatabase();
+ assertThat(restCatalog.namespaceExists(Namespace.of("mydb"))).isTrue();
+ }
+
@Test
public void testToRestLocationNormalisesScheme() {
// s3a:// and legacy s3n:// are rewritten to s3:// (Glue's REST
endpoint only accepts
