This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 74de33af4b [#12237] fix(iceberg): de-flake IcebergClassLoaderPoolIT by
tolerating concurrent JdbcCatalog view-migration on a shared iceberg_tables
(#12240)
74de33af4b is described below
commit 74de33af4bacc423162802b93f4cc3396fd624df
Author: Qi Yu <[email protected]>
AuthorDate: Wed Jul 29 14:22:33 2026 +0800
[#12237] fix(iceberg): de-flake IcebergClassLoaderPoolIT by tolerating
concurrent JdbcCatalog view-migration on a shared iceberg_tables (#12240)
### What changes were proposed in this pull request?
De-flake `IcebergClassLoaderPoolIT` and harden Iceberg JDBC catalog
initialization when several logical catalogs share one JDBC metadata
database:
- Detect the narrow V1 migration conflict where `JdbcCatalog.initialize`
fails because the `iceberg_type` column already exists.
- Close the failed catalog's partially initialized JDBC connection pool
and `FileIO` before discarding it. Iceberg creates these resources
before the schema migration but does not register them with its
`CloseableGroup` until the migration succeeds.
- Initialize a fresh catalog once after the conflict. The shared schema
has already been migrated, so the retry can complete normally.
- Recognize MySQL, PostgreSQL, and SQLite duplicate-column error
wordings in `isConcurrentViewMigrationConflict`.
- Keep the existing access-denied conversion null-safe and simplify the
authentication branch without changing its behavior.
### Why are the changes needed?
Iceberg JDBC catalogs that use the same backend `uri` share the physical
`iceberg_tables` and `iceberg_namespace_properties` control tables.
Different `catalog-backend-name` values isolate their logical namespaces
and tables through the `catalog_name` column, so this is a valid,
although relatively narrow, deployment topology.
Gravitino defaults `jdbc.schema-version` to `V1`. Iceberg creates
`iceberg_tables` without `iceberg_type`, checks for that column with
JDBC metadata, and then runs a non-idempotent migration:
```sql
ALTER TABLE iceberg_tables ADD COLUMN iceberg_type ...
```
The conflict can occur when:
1. Two JDBC catalog instances point to the same backend `uri`.
2. They initialize concurrently while the shared control table is being
migrated to V1, or a JDBC metadata probe temporarily reports the
already-added column as missing.
3. Both instances attempt the same `ALTER TABLE`; the losing instance
fails with a duplicate-column error such as:
```text
Caused by: java.sql.SQLSyntaxErrorException: Duplicate column name
'iceberg_type'
at
org.apache.iceberg.jdbc.JdbcCatalog.executeV1CatalogUpdate(JdbcCatalog.java:276)
at
org.apache.iceberg.jdbc.JdbcCatalog.updateSchemaIfRequired(JdbcCatalog.java:235)
```
`IcebergClassLoaderPoolIT` deliberately creates multiple catalogs on the
same MySQL JDBC URL because it verifies that catalogs with the same
isolation key share a ClassLoader and that dropping one catalog does not
break its sibling. This supported topology exposed the migration race in
`JDK17-deploy-mysql`.
The duplicate-column error means another initializer has completed the
shared migration. This patch therefore closes the losing catalog's
partial resources and retries initialization once with a fresh catalog.
Other SQL failures are still propagated unchanged.
Fix: #12237
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Added classifier tests for MySQL, PostgreSQL, and SQLite
duplicate-column messages, plus unrelated and null-message cases.
- Added a deterministic failed-migration test that simulates the
duplicate `iceberg_type` error and verifies that the JDBC pool,
underlying connection, and `FileIO` are closed.
- `./gradlew :iceberg:iceberg-common:test
:iceberg:iceberg-common:spotlessCheck` — all 64 tests pass.
- `IcebergClassLoaderPoolIT` continues to cover the end-to-end
shared-URI scenario in the deploy/MySQL CI.
---
.../iceberg/common/ClosableJdbcCatalog.java | 28 +++++++-
.../iceberg/common/utils/IcebergCatalogUtil.java | 79 ++++++++++++++++++----
.../iceberg/common/TestClosableJdbcCatalog.java | 54 +++++++++++++++
.../common/utils/TestIcebergCatalogUtil.java | 48 +++++++++++++
4 files changed, 193 insertions(+), 16 deletions(-)
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableJdbcCatalog.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableJdbcCatalog.java
index 6e2dd5c07e..bffedf2c8c 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableJdbcCatalog.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableJdbcCatalog.java
@@ -23,6 +23,7 @@ import java.io.Closeable;
import java.util.Map;
import java.util.function.Function;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.catalog.hadoop.auth.KerberosClient;
import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos;
@@ -73,7 +74,12 @@ public class ClosableJdbcCatalog extends JdbcCatalog
implements Closeable, Suppo
*/
@Override
public void initialize(String inputName, Map<String, String> properties) {
- super.initialize(inputName, properties);
+ try {
+ super.initialize(inputName, properties);
+ } catch (RuntimeException e) {
+ closePartiallyInitializedResources(e);
+ throw e;
+ }
AuthenticationConfig authenticationConfig = new
AuthenticationConfig(properties);
if (authenticationConfig.isKerberosAuth()) {
@@ -105,4 +111,24 @@ public class ClosableJdbcCatalog extends JdbcCatalog
implements Closeable, Suppo
public <R> R doKerberosOperations(Executable<R> executable) throws Throwable
{
return KerberosCatalogUtils.doKerberosOperations(this.properties(),
kerberosClient, executable);
}
+
+ private void closePartiallyInitializedResources(RuntimeException
initializationException) {
+ // JdbcCatalog creates these resources before updating the JDBC schema,
but registers them with
+ // its CloseableGroup only after the update succeeds. Its close() method
therefore cannot clean
+ // them up when initialization fails during the schema update.
+ closePartiallyInitializedResource("connections", initializationException);
+ closePartiallyInitializedResource("io", initializationException);
+ }
+
+ private void closePartiallyInitializedResource(
+ String fieldName, RuntimeException initializationException) {
+ try {
+ Closeable resource = (Closeable) FieldUtils.readField(this, fieldName,
true);
+ if (resource != null) {
+ resource.close();
+ }
+ } catch (Exception closeException) {
+ initializationException.addSuppressed(closeException);
+ }
+ }
}
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
index a5587964e5..466a9de314 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
@@ -57,6 +57,12 @@ public class IcebergCatalogUtil {
private static final Logger LOG =
LoggerFactory.getLogger(IcebergCatalogUtil.class);
+ /**
+ * Column that Iceberg adds to the {@code iceberg_tables} control table in
its V1 view-support
+ * migration (see {@code JdbcUtil} in iceberg-core).
+ */
+ private static final String ICEBERG_TYPE_COLUMN = "iceberg_type";
+
private static InMemoryCatalog loadMemoryCatalog(IcebergConfig
icebergConfig) {
String icebergCatalogName = icebergConfig.getCatalogBackendName();
InMemoryCatalog memoryCatalog = new
MemoryCatalogWithMetadataLocationSupport();
@@ -125,29 +131,72 @@ public class IcebergCatalogUtil {
HdfsConfiguration hdfsConfiguration = new HdfsConfiguration();
properties.forEach(hdfsConfiguration::set);
AuthenticationConfig authenticationConfig = new
AuthenticationConfig(properties);
+ if (authenticationConfig.isKerberosAuth()) {
+ hdfsConfiguration.set(HADOOP_SECURITY_AUTHORIZATION, "true");
+ hdfsConfiguration.set(HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+ } else if (!authenticationConfig.isSimpleAuth()) {
+ throw new UnsupportedOperationException(
+ "Unsupported authentication method: " +
authenticationConfig.getAuthType());
+ }
+
try {
- if (authenticationConfig.isSimpleAuth()) {
- jdbcCatalog.setConf(hdfsConfiguration);
- jdbcCatalog.initialize(icebergCatalogName, properties);
- } else if (authenticationConfig.isKerberosAuth()) {
- hdfsConfiguration.set(HADOOP_SECURITY_AUTHORIZATION, "true");
- hdfsConfiguration.set(HADOOP_SECURITY_AUTHENTICATION, "kerberos");
- jdbcCatalog.setConf(hdfsConfiguration);
- jdbcCatalog.initialize(icebergCatalogName, properties);
- } else {
- throw new UnsupportedOperationException(
- "Unsupported authentication method: " +
authenticationConfig.getAuthType());
- }
+ jdbcCatalog.setConf(hdfsConfiguration);
+ jdbcCatalog.initialize(icebergCatalogName, properties);
} catch (UncheckedSQLException e) {
- if (e.getCause() instanceof SQLException
- && e.getCause().getMessage().contains("Access denied")) {
+ Throwable cause = e.getCause();
+ if (cause instanceof SQLException
+ && cause.getMessage() != null
+ && cause.getMessage().contains("Access denied")) {
throw new ConnectionFailedException(e, e.getMessage());
}
- throw e;
+ if (!isConcurrentViewMigrationConflict(e)) {
+ throw e;
+ }
+ // Iceberg's V1 view-support migration adds the `iceberg_type` column to
the shared
+ // `iceberg_tables` control table with a non-idempotent `ALTER TABLE ...
ADD COLUMN`. When
+ // several Iceberg JDBC catalogs share the same backend `uri` (hence one
`iceberg_tables`), a
+ // losing racer finds the column already added by another catalog and
fails initialization.
+ // The schema is already at V1, so re-initialize a fresh catalog once:
the column now exists,
+ // Iceberg skips the migration, and initialization completes.
+ LOG.info(
+ "iceberg_type column already added by another Iceberg JDBC catalog
sharing the same "
+ + "backend uri; re-initializing catalog {}",
+ icebergCatalogName);
+ jdbcCatalog =
+ new JdbcCatalogWithMetadataLocationSupport(
+ icebergConfig.get(IcebergConfig.JDBC_INIT_TABLES));
+ jdbcCatalog.setConf(hdfsConfiguration);
+ jdbcCatalog.initialize(icebergCatalogName, properties);
}
return jdbcCatalog;
}
+ /**
+ * Whether an {@link UncheckedSQLException} from {@code
JdbcCatalog.initialize} is the benign
+ * conflict raised when another Iceberg JDBC catalog sharing the same
backend {@code uri} already
+ * ran Iceberg's V1 view-support migration.
+ *
+ * <p>That migration adds the {@value #ICEBERG_TYPE_COLUMN} column to the
shared {@code
+ * iceberg_tables} table with a non-idempotent {@code ALTER TABLE ... ADD
COLUMN}, so a losing
+ * racer fails with a duplicate-column error whose wording differs per
database (MySQL: {@code
+ * Duplicate column name 'iceberg_type'}; PostgreSQL: {@code column
"iceberg_type" ... already
+ * exists}; SQLite: {@code duplicate column name: iceberg_type}).
+ *
+ * @param e the exception thrown by catalog initialization
+ * @return {@code true} if the failure is a duplicate {@value
#ICEBERG_TYPE_COLUMN} column
+ * conflict
+ */
+ @VisibleForTesting
+ static boolean isConcurrentViewMigrationConflict(UncheckedSQLException e) {
+ Throwable cause = e.getCause();
+ if (!(cause instanceof SQLException) || cause.getMessage() == null) {
+ return false;
+ }
+ String message = cause.getMessage().toLowerCase(Locale.ROOT);
+ return message.contains(ICEBERG_TYPE_COLUMN)
+ && (message.contains("duplicate column") || message.contains("already
exists"));
+ }
+
private static Catalog loadRestCatalog(IcebergConfig icebergConfig) {
String icebergCatalogName = icebergConfig.getCatalogBackendName();
RESTCatalog restCatalog = new RESTCatalog();
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestClosableJdbcCatalog.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestClosableJdbcCatalog.java
index 4e386fa5fa..f142af2a1b 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestClosableJdbcCatalog.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestClosableJdbcCatalog.java
@@ -20,6 +20,12 @@
package org.apache.gravitino.iceberg.common;
import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLSyntaxErrorException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
@@ -28,9 +34,13 @@ import
org.apache.gravitino.iceberg.common.authentication.kerberos.KerberosConfi
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.jdbc.JdbcClientPool;
+import org.apache.iceberg.jdbc.UncheckedSQLException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
public class TestClosableJdbcCatalog {
@@ -46,6 +56,35 @@ public class TestClosableJdbcCatalog {
Assertions.assertDoesNotThrow(catalog::close);
}
+ @Test
+ void testFailedSchemaMigrationClosesPartiallyCreatedResources() throws
Exception {
+ FileIO fileIO = Mockito.mock(FileIO.class);
+ Connection connection = Mockito.mock(Connection.class);
+ DatabaseMetaData metadata = Mockito.mock(DatabaseMetaData.class);
+ ResultSet columns = Mockito.mock(ResultSet.class);
+ PreparedStatement alterTable = Mockito.mock(PreparedStatement.class);
+ Mockito.when(connection.getMetaData()).thenReturn(metadata);
+ Mockito.when(metadata.getColumns(null, null, "iceberg_tables",
"iceberg_type"))
+ .thenReturn(columns);
+ Mockito.when(columns.next()).thenReturn(false);
+
Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(alterTable);
+ Mockito.when(alterTable.execute())
+ .thenThrow(new SQLSyntaxErrorException("Duplicate column name
'iceberg_type'"));
+
+ JdbcClientPool clientPool = new TestJdbcClientPool(connection);
+ ClosableJdbcCatalog catalog =
+ new ClosableJdbcCatalog(properties -> fileIO, properties ->
clientPool, false);
+ Map<String, String> properties = newJdbcCatalogProperties();
+ properties.put(IcebergConstants.ICEBERG_JDBC_SCHEMA_VERSION, "V1");
+
+ Assertions.assertThrows(
+ UncheckedSQLException.class, () -> catalog.initialize("test",
properties));
+
+ Assertions.assertTrue(clientPool.isClosed());
+ Mockito.verify(connection).close();
+ Mockito.verify(fileIO).close();
+ }
+
@Test
void testKerberosInitNoConf() {
ClosableJdbcCatalog catalog = new ClosableJdbcCatalog();
@@ -131,4 +170,19 @@ public class TestClosableJdbcCatalog {
properties.put(IcebergConstants.ICEBERG_JDBC_INITIALIZE, "true");
return properties;
}
+
+ private static class TestJdbcClientPool extends JdbcClientPool {
+ private final Connection connection;
+
+ private TestJdbcClientPool(Connection connection) {
+ super(1, "jdbc:test", Collections.emptyMap());
+ this.connection = connection;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected Connection newClient() {
+ return connection;
+ }
+ }
}
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
index 1b302f4437..79e18b5028 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
@@ -20,6 +20,8 @@
package org.apache.gravitino.iceberg.common.utils;
import java.nio.file.Path;
+import java.sql.SQLException;
+import java.sql.SQLSyntaxErrorException;
import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
@@ -36,6 +38,7 @@ import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.jdbc.JdbcCatalogWithMetadataLocationSupport;
+import org.apache.iceberg.jdbc.UncheckedSQLException;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -290,4 +293,49 @@ public class TestIcebergCatalogUtil {
Assertions.assertEquals(
"6789",
properties.get(IcebergConstants.ICEBERG_REST_CLIENT_SOCKET_TIMEOUT_MS));
}
+
+ @Test
+ void testIsConcurrentViewMigrationConflictAcrossDatabases() {
+ // The duplicate `iceberg_type` column error wording differs per backend
database; each of these
+ // means another Iceberg JDBC catalog on the same `uri` already ran the V1
view migration.
+ Assertions.assertTrue(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ migrationError("Duplicate column name 'iceberg_type'")),
+ "MySQL duplicate-column error should be treated as a concurrent
view-migration conflict");
+ Assertions.assertTrue(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ migrationError(
+ "ERROR: column \"iceberg_type\" of relation \"iceberg_tables\"
already exists")),
+ "PostgreSQL duplicate-column error should be treated as a conflict");
+ Assertions.assertTrue(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ migrationError("duplicate column name: iceberg_type")),
+ "SQLite duplicate-column error should be treated as a conflict");
+ }
+
+ @Test
+ void testIsConcurrentViewMigrationConflictRejectsUnrelatedErrors() {
+ // A duplicate-column error on a different column is not the view
migration.
+ Assertions.assertFalse(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ migrationError("Duplicate column name 'some_other_column'")));
+ // An `iceberg_type` mention without a duplicate/already-exists signal is
not this conflict.
+ Assertions.assertFalse(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ migrationError("Unknown column 'iceberg_type' in 'field list'")));
+ // A non-SQL cause is never this conflict.
+ Assertions.assertFalse(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ new UncheckedSQLException(
+ new RuntimeException("Duplicate column name 'iceberg_type'"),
"boom")));
+ // A null cause message must not throw.
+ Assertions.assertFalse(
+ IcebergCatalogUtil.isConcurrentViewMigrationConflict(
+ new UncheckedSQLException(new SQLException((String) null),
"boom")));
+ }
+
+ private static UncheckedSQLException migrationError(String causeMessage) {
+ return new UncheckedSQLException(
+ new SQLSyntaxErrorException(causeMessage), "Cannot check and
eventually update SQL schema");
+ }
}
