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 2d4796baad [#13767] improve(core,jdbc): verify JDBC catalog backend
reachable at create (#13769)
2d4796baad is described below
commit 2d4796baad1d63386751f18dff8df8a570f4462a
Author: geyanggang <[email protected]>
AuthorDate: Tue Oct 13 11:03:20 2026 +0800
[#13767] improve(core,jdbc): verify JDBC catalog backend reachable at
create (#13769)
### What changes were proposed in this pull request?
Relational JDBC catalog creation now verifies the backend is reachable,
consistently
across JDBC providers, instead of silently succeeding and failing later
on an unrelated
operation. The check is bounded so it fails fast and reports the failure
to the caller
naming the host.
Three coordinated changes:
1. **Bounded JDBC connect/socket timeout (`catalog-jdbc-common`).**
- Add `jdbc.pool.connect-timeout-ms` to `JdbcConfig` (default `30000`,
positive-value
check), plus a `getConnectTimeoutMs()` getter.
- Add a pure, driver-aware utility `JdbcConnectionTimeouts` that
translates this
canonical milliseconds value into each driver family's connection
parameters, since
the parameter unit differs: MySQL Connector/J
(`com.mysql.cj.jdbc.Driver` /
legacy `com.mysql.jdbc.Driver`, also used by Doris/StarRocks) takes
milliseconds;
PostgreSQL (`org.postgresql.Driver`) takes seconds (ms rounded up,
minimum 1). An
unknown driver family yields no properties, so adding a new provider
never breaks.
Any value the operator already set — as a connection property or a
`jdbc-url` query
parameter — is preserved (defaults, never overrides; the URL check is
case-sensitive
and boundary-anchored).
- `DataSourceUtils.createDBCPDataSource` applies the translated
properties via
`addConnectionProperty`, so they are per-connection bounds, do not
collide with the
unsafe-pool-property guard, and keep the pool lazy.
2. **`gravitino.catalog.verify-on-create` control property (`core`).**
Registered in `BaseCatalogPropertiesMetadata` as a hidden, non-reserved
boolean with
default `true`, with the key constant on `Catalog`. It is a registered
property (so it
is not silently dropped during create validation) that lets an
administrator opt out of
create-time verification to register a catalog whose backend is
temporarily unavailable.
3. **Create-time verification via a `validateOnCreate()` hook (`core` +
`jdbc`).**
- Add `BaseCatalog.validateOnCreate()`, whose default body eagerly
initializes `ops()`,
preserving current behavior for existing opt-ins (e.g. Iceberg REST).
- `CatalogManager.initCatalogWrapper` now calls `validateOnCreate()`
instead of `ops()`
on the existing `shouldValidateOnCreate()` opt-in path (one-line
change).
- `JdbcCatalog` overrides `shouldValidateOnCreate()` (reads
`verify-on-create`,
defaulting to `true`) and `validateOnCreate()` to run a single
`testConnection`
(one `listDatabases` round trip). On failure a
`ConnectionFailedException` propagates,
and the existing create rollback removes the just-persisted catalog
entity.
- `JdbcCatalogOperations.testConnection` enriches the
`ConnectionFailedException`
message with the `host:port` parsed from the `jdbc-url` (same exception
type).
Docs: `docs/jdbc-catalog-connection-validation.md` documents the
create-time verification,
the override, and the bounded connect timeout.
### Why are the changes needed?
Creating a relational JDBC catalog returned success without any check
that the backend was
reachable, and the behavior was not consistent across providers. An
administrator who
mistyped a host, port, or credential got a success response and a
catalog that did not
work; the failure then surfaced later on an unrelated operation.
The cause is that `CatalogManager.createCatalog` performed no connection
check — whether
the backend was touched during create depended entirely on each
connector's `initialize()`
side effects. JDBC catalogs build a lazy DBCP2 connection pool
(`DataSourceUtils.createDBCPDataSource`, with `initialSize` blocked), so
create opened no
connection and never reached the backend. The existing
`BaseCatalog.shouldValidateOnCreate()`
hook only eagerly called `ops()`, which for JDBC still opened no
connection, so even
opting in did not verify reachability.
There was also no bounded JDBC connect/socket timeout: `JdbcConfig`
exposed only pool
settings and `DataSourceUtils` set only the pool `maxWait`, so a
connection attempt against
an unreachable host could block for the driver default (effectively
unbounded for the
MySQL-family default `connectTimeout=0`). Bounding it is a prerequisite
for safe create-time
verification (which runs inside the metalake write lock) and a useful
hardening on its own.
Fix: #13767
### Does this PR introduce _any_ user-facing change?
Yes. Creating a relational JDBC catalog verifies the backend is
reachable by default and returns a connection error naming the host on
failure. Verification uses a dedicated, time-bounded probe connection
(connect + read, plus login for PostgreSQL); the normal connection pool
keeps only a connect timeout so long-running operations are not capped.
New config jdbc.pool.connect-timeout-ms (default 30000, max 2147483647).
Set gravitino.catalog.verify-on-create=false to skip the check. Iceberg
REST / model / generic catalogs are unaffected. Tests: connect-timeout
translation per driver (pool) and connect+read+login (probe); an
accept-but-silent server proves the probe is bounded; testConnection
still rejects reserved/missing-required properties; host extraction
never leaks credentials; connect-timeout upper bound.
### How was this patch tested?
New and extended unit tests:
- `catalog-jdbc-common`:
- `TestJdbcConnectionTimeouts` — MySQL family emits milliseconds;
PostgreSQL converts
ms→seconds (round up, minimum 1); unknown driver emits nothing; an
operator-set
`connectTimeout`/`socketTimeout` is preserved whether set as a config
key or a `jdbc-url`
query parameter (case-sensitive, boundary-anchored); custom-value
propagation; null-URL
safe.
- `TestJdbcConfig` — the new setting's default and override.
- `TestDataSourceValidation` — an unknown driver injects nothing and the
pool still borrows
a connection.
- `core`:
- `TestCatalogManager` — an opted-in catalog whose backend probe fails
propagates
`ConnectionFailedException` and rolls back the persisted entity (store
no longer contains
it, identifier is reusable); a non-opted-in catalog is not verified; the
existing backend
validation test is unchanged.
- `TestCatalog` / `TestCatalogOperations` — simulate a failing
`testConnection` without a
real backend.
Commands run (from the repo root):
- `./gradlew spotlessApply` — SUCCESS.
- `./gradlew :catalogs:catalog-jdbc-common:test :core:test -PskipITs` —
BUILD SUCCESSFUL
(core: 2699 tests, 1 skipped, 0 failed), modules compiled with
checkstyle `-Werror`.
---------
Co-authored-by: Qi Yu <[email protected]>
---
.../integration/test/CatalogClickHouseIT.java | 4 +
.../apache/gravitino/catalog/jdbc/JdbcCatalog.java | 46 +++
.../catalog/jdbc/JdbcCatalogOperations.java | 149 +++++++++-
.../jdbc/JdbcCatalogPropertiesMetadata.java | 19 +-
.../gravitino/catalog/jdbc/config/JdbcConfig.java | 64 ++++
.../catalog/jdbc/utils/DataSourceUtils.java | 6 +
.../catalog/jdbc/utils/JdbcConnectionTimeouts.java | 303 +++++++++++++++++++
.../jdbc/TestJdbcCatalogOperationsProbe.java | 204 +++++++++++++
.../jdbc/TestJdbcCatalogPropertiesMetadata.java | 63 ++++
.../jdbc/utils/TestDataSourceValidation.java | 26 ++
.../catalog/jdbc/utils/TestJdbcConfig.java | 43 +++
.../jdbc/utils/TestJdbcConnectionTimeouts.java | 326 +++++++++++++++++++++
.../mysql/integration/test/CatalogMysqlIT.java | 8 +
.../apache/gravitino/catalog/CatalogManager.java | 55 +++-
.../apache/gravitino/connector/BaseCatalog.java | 15 +
.../java/org/apache/gravitino/TestCatalog.java | 17 ++
.../gravitino/catalog/TestCatalogManager.java | 106 +++++++
.../gravitino/connector/TestCatalogOperations.java | 19 ++
docs/jdbc-catalog-connection-validation.md | 50 ++++
19 files changed, 1509 insertions(+), 14 deletions(-)
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
index 461308cf54..8d7444acb8 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
@@ -3274,6 +3274,10 @@ public class CatalogClickHouseIT extends BaseIT {
String password = CLICKHOUSE_CONTAINER.getPassword();
String wrongPassword = password + "wrong";
catalogProperties.put(JdbcConfig.PASSWORD.getKey(), wrongPassword);
+ // This test deliberately creates the catalog with a wrong password and
then fixes it via
+ // alterCatalog, so it must opt out of create-time backend verification;
otherwise the create
+ // itself would be rejected for the unreachable backend (the behavior
added in #13767).
+ catalogProperties.put(JdbcConfig.VERIFY_ON_CREATE.getKey(), "false");
metalake.createCatalog(
testCatalogName, Catalog.Type.RELATIONAL, provider, "comment",
catalogProperties);
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
index 7ac02124f7..d289aa1e0b 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java
@@ -37,6 +37,8 @@ import org.apache.gravitino.connector.PropertiesMetadata;
import org.apache.gravitino.connector.PropertyEntry;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.credential.JdbcCredential;
+import org.apache.gravitino.exceptions.ConnectionFailedException;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
/** Implementation of an Jdbc catalog in Gravitino. */
public abstract class JdbcCatalog extends BaseCatalog<JdbcCatalog> {
@@ -79,6 +81,50 @@ public abstract class JdbcCatalog extends
BaseCatalog<JdbcCatalog> {
return new JdbcCatalogCapability();
}
+ /**
+ * Returns the effective {@link JdbcConfig#VERIFY_ON_CREATE
verify-on-create} setting for this
+ * catalog. A relational JDBC catalog verifies its backend at create time by
default; an
+ * administrator opts out by setting {@code verify-on-create=false} to
register a catalog whose
+ * backend is temporarily unavailable.
+ *
+ * @return {@code true} to verify the backend at create (default); {@code
false} to skip the check
+ */
+ @Override
+ public boolean shouldValidateOnCreate() {
+ Map<String, String> properties = entity().getProperties();
+ if (properties == null) {
+ return true;
+ }
+ return new JdbcConfig(properties).getVerifyOnCreate();
+ }
+
+ /**
+ * Verifies the backend is reachable at create time with a dedicated,
time-bounded one-off probe
+ * connection (see {@link JdbcCatalogOperations#probeConnection}). On
failure the probe raises a
+ * {@link ConnectionFailedException}, which propagates so {@code
CatalogManager} rolls back the
+ * just-persisted catalog entity.
+ *
+ * <p>The probe does not borrow from the lazy pool: it opens its own
connection bounded by both
+ * the connect and the read/login timeout, so a backend that accepts the
socket but never speaks
+ * the protocol fails fast instead of blocking the metalake write lock. The
probe connection is
+ * opened, validated and closed within the call, so there is no pool or
connection leak.
+ */
+ @Override
+ public void validateOnCreate() {
+ try {
+ ((JdbcCatalogOperations)
ops()).probeConnection(entity().nameIdentifier());
+ } catch (RuntimeException e) {
+ // Includes ConnectionFailedException; rethrow unchanged so
CatalogManager rolls back and the
+ // connection-failed taxonomy the caller relies on is preserved.
+ throw e;
+ } catch (Exception e) {
+ // Defensive: the probe only throws runtime exceptions, so wrap any
unexpected checked
+ // exception without changing the connection-failed taxonomy the caller
relies on.
+ throw new GravitinoRuntimeException(
+ e, "Failed to verify JDBC catalog backend at create: %s",
e.getMessage());
+ }
+ }
+
/**
* @return The {@link JdbcExceptionConverter} to be used by the catalog.
*/
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
index 7807b68134..7d771b7ac2 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogOperations.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.catalog.jdbc;
import static org.apache.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import java.sql.Connection;
@@ -34,6 +35,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Properties;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.sql.DataSource;
@@ -53,6 +56,7 @@ import
org.apache.gravitino.catalog.jdbc.operation.JdbcTableOperations;
import org.apache.gravitino.catalog.jdbc.operation.RequireDatabaseOperation;
import org.apache.gravitino.catalog.jdbc.operation.TableOperation;
import org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
+import org.apache.gravitino.catalog.jdbc.utils.JdbcConnectionTimeouts;
import org.apache.gravitino.connector.CatalogInfo;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HasPropertyMetadata;
@@ -105,6 +109,7 @@ public class JdbcCatalogOperations
private DataSource dataSource;
private String jdbcUrl;
+ private JdbcConfig jdbcConfig;
private final JdbcColumnDefaultValueConverter columnDefaultValueConverter;
@@ -172,7 +177,7 @@ public class JdbcCatalogOperations
Map<String, String> resultConf = Maps.newHashMap(prefixMap);
resultConf.putAll(gravitinoConfig);
- JdbcConfig jdbcConfig = new JdbcConfig(resultConf);
+ this.jdbcConfig = new JdbcConfig(resultConf);
this.jdbcUrl = jdbcConfig.getJdbcUrl();
this.dataSource = DataSourceUtils.createDataSource(jdbcConfig);
@@ -239,21 +244,159 @@ public class JdbcCatalogOperations
*
* @param catalogIdent the name of the catalog.
*/
+ /**
+ * Verifies the backend is reachable using a dedicated, time-bounded one-off
connection rather
+ * than a pooled one. Unlike {@link #testConnection(NameIdentifier)} (which
borrows from the lazy
+ * pool, whose connections carry only a connect timeout so normal operations
stay unbounded), this
+ * probe applies both the connect timeout and the socket-read (and, for
PostgreSQL, login) timeout
+ * from {@link JdbcConnectionTimeouts#probeConnectionProperties}. That
matters because the probe
+ * runs under the metalake write lock during catalog creation: a backend
that accepts the TCP
+ * connection but never speaks the database protocol would otherwise block
forever and starve the
+ * whole metalake. The connection is opened, validated and closed within
this call.
+ *
+ * @param catalogIdent the identifier of the catalog being verified
+ * @throws ConnectionFailedException if the backend cannot be reached within
the bounded timeout
+ */
+ /**
+ * Sets just the configuration the backend probe needs ({@link #jdbcConfig}
and {@link #jdbcUrl}),
+ * without building the connection pool. For tests that exercise {@link
#probeConnection} with a
+ * fake driver and no real database.
+ *
+ * @param config the resolved JDBC configuration
+ */
+ @VisibleForTesting
+ void setProbeConfigForTest(JdbcConfig config) {
+ this.jdbcConfig = config;
+ this.jdbcUrl = config.getJdbcUrl();
+ }
+
+ public void probeConnection(NameIdentifier catalogIdent) {
+ Driver driver;
+ try {
+ driver = getDriver();
+ } catch (SQLException e) {
+ throw new ConnectionFailedException(
+ e,
+ "Failed to connect to JDBC catalog at %s: %s",
+ hostPortFromJdbcUrl(jdbcUrl),
+ e.getMessage());
+ }
+
+ Properties properties = new Properties();
+ String user = jdbcConfig.getUsername();
+ if (user != null) {
+ properties.setProperty("user", user);
+ }
+ String password = jdbcConfig.getPassword();
+ if (password != null) {
+ properties.setProperty("password", password);
+ }
+ // Bound both the connect and the read/login timeout for this probe so an
accept-but-silent
+ // backend fails fast instead of hanging under the metalake write lock.
+ JdbcConnectionTimeouts.probeConnectionProperties(jdbcConfig, jdbcUrl)
+ .forEach(properties::setProperty);
+
+ try (Connection connection = driver.connect(jdbcUrl, properties)) {
+ if (connection == null) {
+ throw new ConnectionFailedException(
+ "Failed to connect to JDBC catalog at %s: the driver did not
accept the URL",
+ hostPortFromJdbcUrl(jdbcUrl));
+ }
+ // A lightweight round trip that confirms the connection is actually
usable. isValid() is
+ // bounded by the same read timeout applied above.
+ connection.getMetaData().getDatabaseProductName();
+ } catch (ConnectionFailedException e) {
+ throw e;
+ } catch (SQLException e) {
+ throw new ConnectionFailedException(
+ e,
+ "Failed to connect to JDBC catalog at %s: %s",
+ hostPortFromJdbcUrl(jdbcUrl),
+ e.getMessage());
+ }
+ }
+
@Override
public void testConnection(NameIdentifier catalogIdent) {
try {
databaseOperation.listDatabases();
} catch (ConnectionFailedException e) {
- throw e;
+ // Already classified as a connection failure (for example a MySQL
access-denied or a
+ // PostgreSQL SQLSTATE 08 error). Re-wrap it with the host context so
every connection failure
+ // satisfies the documented "names the host" contract, preserving the
original cause.
+ throw new ConnectionFailedException(
+ e.getCause() != null ? e.getCause() : e,
+ "Failed to connect to JDBC catalog at %s: %s",
+ hostPortFromJdbcUrl(jdbcUrl),
+ e.getMessage());
} catch (GravitinoRuntimeException e) {
if (e.getClass() == GravitinoRuntimeException.class && e.getCause()
instanceof SQLException) {
throw new ConnectionFailedException(
- e.getCause(), "Failed to connect to JDBC catalog: %s",
e.getMessage());
+ e.getCause(),
+ "Failed to connect to JDBC catalog at %s: %s",
+ hostPortFromJdbcUrl(jdbcUrl),
+ e.getMessage());
}
throw e;
}
}
+ /**
+ * A plain authority: an optional {@code user-info@}, then either a
bracketed IPv6 literal or a
+ * host of letters/digits/dots/hyphens, then an optional {@code :port}.
Deliberately strict so
+ * that credential-bearing forms the driver also accepts — MySQL's key-value
authority {@code
+ * (host=..,password=..)} and its {@code address=(host=..)(password=..)}
form — do NOT match and
+ * are reported as {@code "unknown"} instead of leaking the password.
+ */
+ private static final Pattern HOST_PORT_AUTHORITY =
+
Pattern.compile("^(?:[^@/]*@)?(\\[[0-9A-Fa-f:]+\\]|[A-Za-z0-9.\\-]+)(?::\\d+)?$");
+
+ /**
+ * Extracts {@code host:port} from a {@code
jdbc:subprotocol://host:port/...} URL for use in a
+ * connection-failure message. Returns {@code "unknown"} whenever the
authority is not a plain
+ * {@code host[:port]} (including a URL with no {@code ://} authority, or
MySQL key-value / {@code
+ * address=(...)} host syntaxes) rather than the raw URL, because the raw
URL and those
+ * alternative syntaxes can carry credentials (for example {@code
jdbc:postgresql:db?password=...}
+ * or {@code jdbc:mysql://(host=h,password=s)/db}) that must not leak into
the exception, the REST
+ * response, or the server logs. Any user-info is stripped. Never throws, so
message formatting
+ * cannot mask the original connection failure.
+ *
+ * @param url the JDBC URL stored for this catalog
+ * @return a {@code host:port} (or {@code host}) description, or {@code
"unknown"} when it is not
+ * a plain host authority
+ */
+ static String hostPortFromJdbcUrl(String url) {
+ if (StringUtils.isBlank(url)) {
+ return "unknown";
+ }
+ int schemeEnd = url.indexOf("://");
+ if (schemeEnd < 0) {
+ return "unknown";
+ }
+ int authorityStart = schemeEnd + "://".length();
+ // The authority ends at the first '/', '?', or ';' following it.
+ int authorityEnd = url.length();
+ for (int i = authorityStart; i < url.length(); i++) {
+ char c = url.charAt(i);
+ if (c == '/' || c == '?' || c == ';') {
+ authorityEnd = i;
+ break;
+ }
+ }
+ String authority = url.substring(authorityStart, authorityEnd);
+ // Only accept a plain host[:port] authority; anything else (blank, or a
credential-bearing
+ // key-value / address=(...) form) is reported as unknown rather than
risking a leak.
+ if (!HOST_PORT_AUTHORITY.matcher(authority).matches()) {
+ return "unknown";
+ }
+ // Strip any user-info ("user@host:port"); keep only the host:port part.
+ int at = authority.lastIndexOf('@');
+ if (at >= 0) {
+ authority = authority.substring(at + 1);
+ }
+ return StringUtils.isBlank(authority) ? "unknown" : authority;
+ }
+
/**
* Creates a new schema with the provided identifier, comment and metadata.
*
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
index 062dd43888..3f933b0975 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.catalog.jdbc;
import static
org.apache.gravitino.connector.PropertyEntry.booleanPropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.integerPropertyEntry;
+import static
org.apache.gravitino.connector.PropertyEntry.longOptionalPropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
import static org.apache.gravitino.connector.PropertyEntry.stringPropertyEntry;
@@ -46,7 +47,9 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
JdbcConfig.POOL_MIN_SIZE.getKey(),
JdbcConfig.POOL_MAX_SIZE.getKey(),
JdbcConfig.POOL_MAX_IDLE.getKey(),
- JdbcConfig.TEST_ON_BORROW.getKey());
+ JdbcConfig.TEST_ON_BORROW.getKey(),
+ JdbcConfig.CONNECT_TIMEOUT_MS.getKey(),
+ JdbcConfig.VERIFY_ON_CREATE.getKey());
static {
List<PropertyEntry<?>> propertyEntries =
@@ -120,6 +123,20 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
false /* immutable */,
JdbcConfig.TEST_ON_BORROW.getDefaultValue(),
true /* hidden */,
+ false /* reserved */),
+ longOptionalPropertyEntry(
+ JdbcConfig.CONNECT_TIMEOUT_MS.getKey(),
+ JdbcConfig.CONNECT_TIMEOUT_MS.getDoc(),
+ false /* immutable */,
+ JdbcConfig.CONNECT_TIMEOUT_MS.getDefaultValue(),
+ false /* hidden */),
+ booleanPropertyEntry(
+ JdbcConfig.VERIFY_ON_CREATE.getKey(),
+ JdbcConfig.VERIFY_ON_CREATE.getDoc(),
+ false /* required */,
+ false /* immutable */,
+ JdbcConfig.VERIFY_ON_CREATE.getDefaultValue(),
+ true /* hidden */,
false /* reserved */));
PROPERTIES_METADATA =
ImmutableMap.<String, PropertyEntry<?>>builder()
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
index 753a9fac84..d6426e6f54 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/config/JdbcConfig.java
@@ -114,6 +114,52 @@ public class JdbcConfig extends Config {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(30000L);
+ /**
+ * The canonical connect/read timeout, in milliseconds, used to bound a JDBC
connection so an
+ * unreachable or unresponsive backend fails fast instead of hanging. It is
translated to each
+ * driver family's own parameters (and unit) by {@link
+ * org.apache.gravitino.catalog.jdbc.utils.JdbcConnectionTimeouts}.
+ *
+ * <p>It is applied differently to the two kinds of connection: the pooled
connections used by
+ * normal operations get only the <em>connect</em> timeout (a read timeout
there would abort
+ * legitimate long-running operations), while the one-off create-time
verification probe gets both
+ * the <em>connect</em> and the <em>read/login</em> timeout, because that
probe runs under the
+ * metalake write lock and a backend that accepts the socket but never
speaks the protocol would
+ * otherwise block the connect-only bound forever. An operator value set
directly on the
+ * connection or in the {@code jdbc-url} is never overridden.
+ *
+ * <p>The value is bounded by {@link Integer#MAX_VALUE} milliseconds because
the MySQL and
+ * PostgreSQL driver timeouts are {@code int}s (and the PostgreSQL
millisecond-to-seconds
+ * conversion stays within {@code int} range); a larger value would be
accepted here and only fail
+ * later when the driver parses it.
+ */
+ public static final ConfigEntry<Long> CONNECT_TIMEOUT_MS =
+ new ConfigBuilder("jdbc.pool.connect-timeout-ms")
+ .doc(
+ "The connect/read timeout for a JDBC connection, in milliseconds
(1.."
+ + Integer.MAX_VALUE
+ + "), translated per driver so an unreachable or silent
backend fails fast")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .longConf()
+ .checkValue(
+ value -> value > 0 && value <= Integer.MAX_VALUE,
+ "The value must be a positive number not greater than " +
Integer.MAX_VALUE)
+ .createWithDefault(30000L);
+
+ /**
+ * Whether a relational JDBC catalog verifies that its backend is reachable
at create time. When
+ * {@code true} (the default, and when absent) the catalog runs a bounded
connection probe during
+ * create and the create fails if the backend is unreachable; set it to
{@code false} to register
+ * a catalog whose backend is temporarily unavailable. It is a catalog
property, not a connection
+ * setting, so it is not forwarded to the driver.
+ */
+ public static final ConfigEntry<Boolean> VERIFY_ON_CREATE =
+ new ConfigBuilder("verify-on-create")
+ .doc("Whether a relational JDBC catalog verifies its backend is
reachable at create time")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .booleanConf()
+ .createWithDefault(true);
+
public String getJdbcUrl() {
return get(JDBC_URL);
}
@@ -159,6 +205,24 @@ public class JdbcConfig extends Config {
return get(POOL_MAX_WAIT_MS);
}
+ /**
+ * Returns the canonical connect/socket timeout for a single JDBC
connection, in milliseconds.
+ *
+ * @return the connect/socket timeout in milliseconds
+ */
+ public long getConnectTimeoutMs() {
+ return get(CONNECT_TIMEOUT_MS);
+ }
+
+ /**
+ * Returns whether this catalog verifies that its backend is reachable at
create time.
+ *
+ * @return {@code true} to verify the backend at create (default); {@code
false} to skip the check
+ */
+ public boolean getVerifyOnCreate() {
+ return get(VERIFY_ON_CREATE);
+ }
+
public JdbcConfig(Map<String, String> properties) {
super(false);
loadFromMap(properties, k -> true);
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
index 40f4fc90dd..c2d224b7a1 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
@@ -167,6 +167,12 @@ public class DataSourceUtils {
// Validate connections on borrow when enabled.
basicDataSource.setTestOnBorrow(jdbcConfig.getTestOnBorrow());
basicDataSource.setMaxWait(Duration.ofMillis(jdbcConfig.getMaxWaitMs()));
+ // Apply the driver-aware connect/socket timeout as DBCP connection
properties (not pool keys,
+ // so no collision with UNSAFE_POOL_PROPERTIES and the pool stays lazy).
These bound a single
+ // connection and take effect on the first real connection — i.e.
create-time verification — so
+ // an unreachable backend fails fast instead of hanging. Operator-set
values are preserved.
+ JdbcConnectionTimeouts.connectionProperties(jdbcConfig, jdbcUrl)
+ .forEach(basicDataSource::addConnectionProperty);
return basicDataSource;
}
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
new file mode 100644
index 0000000000..88188551f0
--- /dev/null
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/JdbcConnectionTimeouts.java
@@ -0,0 +1,303 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.jdbc.utils;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Translates the canonical {@link JdbcConfig#CONNECT_TIMEOUT_MS
connect-timeout-ms} into the
+ * driver-family connection properties that bound a JDBC connection, so an
unreachable or silent
+ * backend fails fast instead of hanging.
+ *
+ * <p>Two translations are provided, because a pooled connection and the
create-time verification
+ * probe need different bounds:
+ *
+ * <ul>
+ * <li>{@link #connectionProperties} — used for the <b>pooled</b>
connections of normal
+ * operations. It bounds only the <em>connection-establishment</em>
timeout, deliberately not
+ * the socket-read timeout: a read timeout on the pool would also abort
legitimate
+ * long-running operations on a healthy backend (a large {@code ALTER
TABLE}, a slow metadata
+ * query).
+ * <li>{@link #probeConnectionProperties} — used for the one-off
<b>create-time verification
+ * probe</b>. It bounds the connect timeout <em>and</em> the socket-read
(and, for PostgreSQL,
+ * login) timeout, because that probe runs under the metalake write
lock: a backend that
+ * accepts the TCP connection but never speaks the database protocol (a
load balancer with a
+ * dead backend, a wrong port, a proxy that accepts and never forwards)
would otherwise make
+ * the connect-only bound wait on the driver's default read timeout,
which is "wait forever",
+ * and hold the lock for the whole metalake.
+ * </ul>
+ *
+ * <p>DBCP has no portable timeout; the real bound lives in the driver, and
both the parameter name
+ * and its unit differ per driver family:
+ *
+ * <ul>
+ * <li><b>MySQL Connector/J</b> ({@code com.mysql.cj.jdbc.Driver}, the
legacy {@code
+ * com.mysql.jdbc.Driver} — also used by Doris and StarRocks — and the
MySQL-compatible
+ * OceanBase Connector/J {@code com.oceanbase.jdbc.Driver}): {@code
connectTimeout} and {@code
+ * socketTimeout} in <b>milliseconds</b>.
+ * <li><b>PostgreSQL</b> ({@code org.postgresql.Driver}, also used by the
PostgreSQL-compatible
+ * Hologres catalog): {@code connectTimeout}, {@code socketTimeout} and
{@code loginTimeout}
+ * in <b>seconds</b>.
+ * <li><b>ClickHouse</b> ({@code com.clickhouse.jdbc.ClickHouseDriver}):
{@code connect_timeout}
+ * and {@code socket_timeout} in <b>milliseconds</b> (note the
underscore names).
+ * </ul>
+ *
+ * <p>An unknown driver family yields no properties (logged at debug), so
adding a new JDBC provider
+ * never breaks; that provider opts in by registering its family here.
+ *
+ * <p><b>Default, never override.</b> A property is omitted when the operator
has already set it in
+ * one of the two places that actually reach the driver: a query parameter in
the {@code jdbc-url},
+ * or inside the DBCP {@code connectionProperties} value (a {@code
name=value;...} string), so the
+ * operator's value always wins. Both checks are case-sensitive and anchored
at a boundary ({@code
+ * ?} or {@code &} for the URL, {@code ;} for connectionProperties), so a
differently-cased or
+ * substring variant does not suppress injection of the canonical key. A bare
top-level config key
+ * of the same name is deliberately not treated as operator-set, because DBCP
does not honor such a
+ * key.
+ *
+ * <p>The pooled properties are applied as DBCP <em>connection</em> properties
(via {@code
+ * BasicDataSource.addConnectionProperty}), not raw pool keys, so they neither
collide with the
+ * unsafe pool properties nor make the pool eager.
+ */
+public final class JdbcConnectionTimeouts {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcConnectionTimeouts.class);
+
+ /** The MySQL Connector/J driver class (modern). */
+ static final String MYSQL_DRIVER = "com.mysql.cj.jdbc.Driver";
+
+ /** The MySQL Connector/J driver class (legacy, pre-8.0). */
+ static final String MYSQL_LEGACY_DRIVER = "com.mysql.jdbc.Driver";
+
+ /** The OceanBase Connector/J driver class (MySQL-compatible, milliseconds).
*/
+ static final String OCEANBASE_DRIVER = "com.oceanbase.jdbc.Driver";
+
+ /** The PostgreSQL driver class (also used by the Hologres catalog). */
+ static final String POSTGRESQL_DRIVER = "org.postgresql.Driver";
+
+ /** The ClickHouse JDBC driver class. */
+ static final String CLICKHOUSE_DRIVER =
"com.clickhouse.jdbc.ClickHouseDriver";
+
+ /** The connect-timeout parameter name used by the MySQL and PostgreSQL
families (camelCase). */
+ static final String CONNECT_TIMEOUT_PARAM = "connectTimeout";
+
+ /** The socket-read-timeout parameter name used by the MySQL and PostgreSQL
families. */
+ static final String SOCKET_TIMEOUT_PARAM = "socketTimeout";
+
+ /** The login-timeout parameter name used by the PostgreSQL family. */
+ static final String LOGIN_TIMEOUT_PARAM = "loginTimeout";
+
+ /** The connect-timeout parameter name used by ClickHouse (snake_case). */
+ static final String CLICKHOUSE_CONNECT_TIMEOUT_PARAM = "connect_timeout";
+
+ /** The socket-read-timeout parameter name used by ClickHouse (snake_case).
*/
+ static final String CLICKHOUSE_SOCKET_TIMEOUT_PARAM = "socket_timeout";
+
+ private JdbcConnectionTimeouts() {
+ // Utility class, prevent instantiation.
+ }
+
+ /**
+ * Returns the driver-family <b>connect-timeout</b> connection property that
bounds a pooled
+ * connection to the canonical {@link JdbcConfig#getConnectTimeoutMs()}
value, or an empty map
+ * when the driver family is unknown or the operator has already set the
connect timeout. Only the
+ * connect timeout is returned here; see {@link #probeConnectionProperties}
for the create-time
+ * probe, which also bounds the read timeout.
+ *
+ * @param config the catalog JDBC configuration; must not be null
+ * @param jdbcUrl the JDBC URL whose query parameters are inspected for
operator-set timeouts; may
+ * be null or blank
+ * @return a map of connect-timeout property name to value, or an empty map
+ */
+ public static Map<String, String> connectionProperties(JdbcConfig config,
String jdbcUrl) {
+ return timeoutProperties(config, jdbcUrl, false /* includeReadTimeout */);
+ }
+
+ /**
+ * Returns the driver-family <b>connect and read (and, for PostgreSQL,
login)</b> timeout
+ * connection properties that bound the one-off create-time verification
probe to the canonical
+ * {@link JdbcConfig#getConnectTimeoutMs()} value, or an empty map when the
driver family is
+ * unknown or the operator has already set the properties. Bounding the read
timeout here (but not
+ * on the pool, see {@link #connectionProperties}) ensures a backend that
accepts the socket but
+ * never speaks the protocol fails the probe instead of blocking the
metalake write lock forever.
+ *
+ * @param config the catalog JDBC configuration; must not be null
+ * @param jdbcUrl the JDBC URL whose query parameters are inspected for
operator-set timeouts; may
+ * be null or blank
+ * @return a map of timeout property name to value, or an empty map
+ */
+ public static Map<String, String> probeConnectionProperties(JdbcConfig
config, String jdbcUrl) {
+ return timeoutProperties(config, jdbcUrl, true /* includeReadTimeout */);
+ }
+
+ private static Map<String, String> timeoutProperties(
+ JdbcConfig config, String jdbcUrl, boolean includeReadTimeout) {
+ Map<String, String> properties = new LinkedHashMap<>();
+ String driver = config.getJdbcDriver();
+ long millis = config.getConnectTimeoutMs();
+
+ String connectParam;
+ String readParam;
+ String loginParam;
+ boolean seconds;
+ if (MYSQL_DRIVER.equals(driver)
+ || MYSQL_LEGACY_DRIVER.equals(driver)
+ || OCEANBASE_DRIVER.equals(driver)) {
+ // MySQL Connector/J and the MySQL-compatible OceanBase driver expect
milliseconds.
+ connectParam = CONNECT_TIMEOUT_PARAM;
+ readParam = SOCKET_TIMEOUT_PARAM;
+ loginParam = null;
+ seconds = false;
+ } else if (POSTGRESQL_DRIVER.equals(driver)) {
+ // PostgreSQL (and Hologres via the PostgreSQL driver) expect seconds.
loginTimeout bounds the
+ // authentication handshake, which socketTimeout alone does not fully
cover.
+ connectParam = CONNECT_TIMEOUT_PARAM;
+ readParam = SOCKET_TIMEOUT_PARAM;
+ loginParam = LOGIN_TIMEOUT_PARAM;
+ seconds = true;
+ } else if (CLICKHOUSE_DRIVER.equals(driver)) {
+ // ClickHouse uses snake_case parameter names and expects milliseconds.
+ connectParam = CLICKHOUSE_CONNECT_TIMEOUT_PARAM;
+ readParam = CLICKHOUSE_SOCKET_TIMEOUT_PARAM;
+ loginParam = null;
+ seconds = false;
+ } else {
+ LOG.debug(
+ "JDBC driver '{}' has no known timeout translation; leaving the
connection timeout "
+ + "unbounded by Gravitino. Set it in the jdbc-url if needed.",
+ driver);
+ return properties;
+ }
+
+ String value = Long.toString(seconds ? millisToSeconds(millis) : millis);
+ addIfAbsent(properties, connectParam, value, config, jdbcUrl);
+ if (includeReadTimeout) {
+ addIfAbsent(properties, readParam, value, config, jdbcUrl);
+ if (loginParam != null) {
+ addIfAbsent(properties, loginParam, value, config, jdbcUrl);
+ }
+ }
+ return properties;
+ }
+
+ /**
+ * Converts milliseconds to seconds, rounded up with a minimum of 1 so a
sub-second bound never
+ * collapses to 0 ("no timeout"). The arithmetic is overflow-safe for the
whole positive {@code
+ * long} range (the input is validated positive and {@code <=
Integer.MAX_VALUE} by {@link
+ * JdbcConfig}).
+ *
+ * @param millis the timeout in milliseconds (positive)
+ * @return the timeout in whole seconds, at least 1
+ */
+ private static long millisToSeconds(long millis) {
+ // (millis - 1) / 1000 + 1 is ceil(millis/1000) for positive millis
without the overflow that
+ // (millis + 999) would hit near Long.MAX_VALUE.
+ return (millis - 1L) / 1000L + 1L;
+ }
+
+ /**
+ * The DBCP2 pool key whose value is a {@code name=value;name2=value2}
string of connection
+ * properties. {@code DataSourceUtils} forwards it to {@code
BasicDataSourceFactory}, which parses
+ * it into the pool's connection properties, and the create-time probe reads
the same catalog
+ * configuration. An operator-set timeout lives inside this value, not as a
top-level config key.
+ */
+ static final String DBCP_CONNECTION_PROPERTIES_KEY = "connectionProperties";
+
+ /**
+ * Adds a timeout property unless the operator has already set it,
preserving the operator's value
+ * in that case. The operator can set it in two places that actually reach
the driver: a {@code
+ * jdbc-url} query parameter, or inside the DBCP {@code
connectionProperties} value (a {@code
+ * name=value;...} string). A bare top-level config key of the same name is
deliberately NOT
+ * treated as operator-set, because DBCP does not recognize such a key —
honoring it would
+ * suppress the bound while the driver still saw no timeout.
+ *
+ * @param properties the map being built
+ * @param name the timeout property name
+ * @param value the Gravitino default value
+ * @param config the catalog configuration
+ * @param jdbcUrl the JDBC URL
+ */
+ private static void addIfAbsent(
+ Map<String, String> properties,
+ String name,
+ String value,
+ JdbcConfig config,
+ String jdbcUrl) {
+ if (hasUrlParameter(jdbcUrl, name)
+ ||
hasConnectionProperty(config.getAllConfig().get(DBCP_CONNECTION_PROPERTIES_KEY),
name)) {
+ return;
+ }
+ properties.put(name, value);
+ }
+
+ /**
+ * Determines whether the DBCP {@code connectionProperties} value already
defines the given
+ * property. The value is a {@code name=value} list separated by {@code ;}
(the format DBCP's
+ * {@code BasicDataSourceFactory} parses), so the name is matched
case-sensitively at a {@code ;}
+ * boundary, mirroring the URL check; a differently-cased or substring
variant does not count.
+ *
+ * @param connectionProperties the DBCP connectionProperties string; may be
null or blank
+ * @param name the exact property name to look for
+ * @return true if the property is present with exactly this name
+ */
+ private static boolean hasConnectionProperty(String connectionProperties,
String name) {
+ if (StringUtils.isBlank(connectionProperties)) {
+ return false;
+ }
+ for (String pair : connectionProperties.split(";")) {
+ int eq = pair.indexOf('=');
+ String key = (eq >= 0 ? pair.substring(0, eq) : pair).trim();
+ if (key.equals(name)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Determines whether the URL already carries the given query parameter.
+ *
+ * <p>The match is case-sensitive and anchored at a parameter boundary
({@code ?} or {@code &}),
+ * so a differently-cased variant is treated as a different parameter and
{@code connectTimeout}
+ * is not considered present merely because {@code tunnelConnectTimeout} is.
+ *
+ * @param url the JDBC URL; may be null or blank
+ * @param parameter the exact query parameter name to look for
+ * @return true if the parameter is already present with exactly this key
+ */
+ private static boolean hasUrlParameter(String url, String parameter) {
+ if (StringUtils.isBlank(url)) {
+ return false;
+ }
+ String needle = parameter + "=";
+ int index = url.indexOf(needle);
+ while (index >= 0) {
+ char preceding = index == 0 ? '\0' : url.charAt(index - 1);
+ if (preceding == '?' || preceding == '&') {
+ return true;
+ }
+ index = url.indexOf(needle, index + 1);
+ }
+ return false;
+ }
+}
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
new file mode 100644
index 0000000000..68cde50c87
--- /dev/null
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogOperationsProbe.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.jdbc;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.Properties;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
+import
org.apache.gravitino.catalog.jdbc.converter.SqliteColumnDefaultValueConverter;
+import org.apache.gravitino.catalog.jdbc.converter.SqliteExceptionConverter;
+import org.apache.gravitino.catalog.jdbc.converter.SqliteTypeConverter;
+import org.apache.gravitino.catalog.jdbc.operation.SqliteDatabaseOperations;
+import org.apache.gravitino.catalog.jdbc.operation.SqliteTableOperations;
+import org.apache.gravitino.exceptions.ConnectionFailedException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for the create-time backend probe and the credential-safe host
extraction in {@link
+ * JdbcCatalogOperations}, exercised without a real database.
+ */
+public class TestJdbcCatalogOperationsProbe {
+
+ /**
+ * A probe against a server that accepts the TCP connection but never
replies must fail within the
+ * configured read timeout, not hang. This is the
metalake-write-lock-starvation case: before the
+ * read timeout was injected on the probe, the driver's handshake read
waited forever.
+ */
+ @Test
+ public void testProbeAgainstAcceptButSilentServerIsBounded() throws
Exception {
+ try (ServerSocket server = new ServerSocket(0)) {
+ server.setSoTimeout(0);
+ // Accept and then stay silent on a background thread, mimicking a
dead-backend load balancer.
+ Thread acceptor =
+ new Thread(
+ () -> {
+ try {
+ Socket socket = server.accept();
+ // Hold the socket open without ever writing a protocol
handshake.
+ Thread.sleep(60_000L);
+ socket.close();
+ } catch (Exception ignored) {
+ // Test is finishing; ignore.
+ }
+ });
+ acceptor.setDaemon(true);
+ acceptor.start();
+
+ int port = server.getLocalPort();
+ // Driver string is a MySQL family name so JdbcConnectionTimeouts
injects connectTimeout +
+ // socketTimeout (ms); the actual driver is the socket-reading fake
below.
+ JdbcCatalogOperations ops = new ProbeOnlyJdbcCatalogOperations(new
SocketReadingDriver());
+ java.util.Map<String, String> config = new java.util.HashMap<>();
+ config.put(JdbcConfig.JDBC_DRIVER.getKey(), "com.mysql.cj.jdbc.Driver");
+ config.put(JdbcConfig.JDBC_URL.getKey(), "jdbc:faketimeout://127.0.0.1:"
+ port + "/db");
+ config.put(JdbcConfig.USERNAME.getKey(), "u");
+ config.put(JdbcConfig.PASSWORD.getKey(), "p");
+ // 1 second probe budget; the test asserts the probe returns well within
that order.
+ config.put(JdbcConfig.CONNECT_TIMEOUT_MS.getKey(), "1000");
+ ops.setProbeConfigForTest(new JdbcConfig(config));
+
+ // The probe must fail fast (bounded by the injected socketTimeout), not
hang on the read.
+ Assertions.assertTimeoutPreemptively(
+ java.time.Duration.ofSeconds(15),
+ () ->
+ Assertions.assertThrows(
+ ConnectionFailedException.class,
+ () -> ops.probeConnection(NameIdentifier.of("m", "c"))));
+ }
+ }
+
+ @Test
+ public void testHostPortFromJdbcUrlNeverLeaksCredentials() {
+ // Plain host[:port] forms are returned.
+ Assertions.assertEquals(
+ "db:3306",
JdbcCatalogOperations.hostPortFromJdbcUrl("jdbc:mysql://db:3306/test"));
+ Assertions.assertEquals(
+ "db",
JdbcCatalogOperations.hostPortFromJdbcUrl("jdbc:mysql://db/test"));
+ Assertions.assertEquals(
+ "db:3306",
+
JdbcCatalogOperations.hostPortFromJdbcUrl("jdbc:mysql://root:s3cret@db:3306/test"));
+ Assertions.assertEquals(
+ "[::1]:5432",
+
JdbcCatalogOperations.hostPortFromJdbcUrl("jdbc:postgresql://[::1]:5432/test"));
+
+ // Credential-bearing or non-plain authorities must be reported as
unknown, never echoed.
+ Assertions.assertEquals(
+ "unknown",
+ JdbcCatalogOperations.hostPortFromJdbcUrl(
+
"jdbc:mysql://(host=db1,port=3306,user=root,password=s3cret)/test"));
+ Assertions.assertEquals(
+ "unknown",
+ JdbcCatalogOperations.hostPortFromJdbcUrl(
+
"jdbc:mysql://address=(host=db1)(port=3306)(password=s3cret)/test"));
+ // No "://" authority (the PostgreSQL short form can carry a password
query parameter).
+ Assertions.assertEquals(
+ "unknown",
JdbcCatalogOperations.hostPortFromJdbcUrl("jdbc:postgresql:db?password=s3cret"));
+ Assertions.assertEquals("unknown",
JdbcCatalogOperations.hostPortFromJdbcUrl(""));
+ Assertions.assertEquals("unknown",
JdbcCatalogOperations.hostPortFromJdbcUrl(null));
+ }
+
+ /** A JdbcCatalogOperations that only needs enough wiring to run the probe
in a unit test. */
+ private static class ProbeOnlyJdbcCatalogOperations extends
JdbcCatalogOperations {
+ private final Driver driver;
+
+ ProbeOnlyJdbcCatalogOperations(Driver driver) {
+ super(
+ new SqliteExceptionConverter(),
+ new SqliteTypeConverter(),
+ new SqliteDatabaseOperations(":memory:"),
+ new SqliteTableOperations(),
+ new SqliteColumnDefaultValueConverter());
+ this.driver = driver;
+ }
+
+ @Override
+ protected Driver getDriver() {
+ return driver;
+ }
+ }
+
+ /**
+ * A fake driver that performs a real blocking socket read against the URL's
host:port, honoring a
+ * {@code socketTimeout} connection property (milliseconds) exactly as a
real driver would. With
+ * the property set the read is bounded and surfaces as a SQLException;
without it the read
+ * blocks.
+ */
+ private static class SocketReadingDriver implements Driver {
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException
{
+ String hostPort = url.substring("jdbc:faketimeout://".length());
+ String host = hostPort.substring(0, hostPort.indexOf(':'));
+ String rest = hostPort.substring(hostPort.indexOf(':') + 1);
+ int port = Integer.parseInt(rest.substring(0, rest.indexOf('/')));
+ int socketTimeoutMs =
+ info.getProperty("socketTimeout") == null
+ ? 0
+ : Integer.parseInt(info.getProperty("socketTimeout"));
+ try (Socket socket = new Socket(host, port)) {
+ socket.setSoTimeout(socketTimeoutMs);
+ // Mimic a protocol handshake read; a silent server makes this block
until socketTimeout.
+ socket.getInputStream().read();
+ } catch (IOException e) {
+ throw new SQLException("handshake read failed: " + e.getMessage(), e);
+ }
+ throw new SQLException("unexpected: silent server returned data");
+ }
+
+ @Override
+ public boolean acceptsURL(String url) {
+ return url != null && url.startsWith("jdbc:faketimeout://");
+ }
+
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
+ return new DriverPropertyInfo[0];
+ }
+
+ @Override
+ public int getMajorVersion() {
+ return 1;
+ }
+
+ @Override
+ public int getMinorVersion() {
+ return 0;
+ }
+
+ @Override
+ public boolean jdbcCompliant() {
+ return false;
+ }
+
+ @Override
+ public java.util.logging.Logger getParentLogger() throws
SQLFeatureNotSupportedException {
+ // This fake driver does not use java.util.logging. FQN is used because
java.sql.Driver
+ // mandates java.util.logging.Logger, whose simple name would clash with
org.slf4j.Logger.
+ throw new SQLFeatureNotSupportedException();
+ }
+ }
+}
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
index 6aa9f1b04c..b6fd8a56bf 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
@@ -54,6 +54,44 @@ public class TestJdbcCatalogPropertiesMetadata {
Assertions.assertFalse(propertiesMetadata.isHiddenProperty(JdbcConfig.POOL_MAX_IDLE.getKey()));
}
+ @Test
+ public void testShouldValidateOnCreateHonorsVerifyOnCreate() {
+ // Default (property absent): verify on create.
+
Assertions.assertTrue(buildCatalog(ImmutableMap.of()).shouldValidateOnCreate());
+ // Explicit true: verify.
+ Assertions.assertTrue(
+ buildCatalog(ImmutableMap.of(JdbcConfig.VERIFY_ON_CREATE.getKey(),
"true"))
+ .shouldValidateOnCreate());
+ // Explicit false: skip.
+ Assertions.assertFalse(
+ buildCatalog(ImmutableMap.of(JdbcConfig.VERIFY_ON_CREATE.getKey(),
"false"))
+ .shouldValidateOnCreate());
+ }
+
+ private static JdbcCatalog buildCatalog(Map<String, String> extraProperties)
{
+ Map<String, String> properties =
+ ImmutableMap.<String, String>builder()
+ .put(JdbcConfig.JDBC_URL.getKey(), "jdbc:sqlite::memory:")
+ .put(JdbcConfig.JDBC_DRIVER.getKey(), "org.sqlite.JDBC")
+ .put(JdbcConfig.USERNAME.getKey(), "user")
+ .put(JdbcConfig.PASSWORD.getKey(), "password")
+ .putAll(extraProperties)
+ .build();
+ CatalogEntity entity =
+ CatalogEntity.builder()
+ .withId(1L)
+ .withName("jdbc_catalog")
+ .withNamespace(Namespace.of("metalake"))
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("jdbc-test")
+ .withComment("test")
+ .withProperties(properties)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ return new
TestJdbcCatalog().withCatalogConf(entity.getProperties()).withCatalogEntity(entity);
+ }
+
@Test
public void testCatalogPropertiesKeepPoolSizeAfterLoad() {
Map<String, String> createProperties =
@@ -96,6 +134,31 @@ public class TestJdbcCatalogPropertiesMetadata {
.get(JdbcConfig.POOL_MAX_IDLE.getKey()));
}
+ @Test
+ public void testConnectTimeoutSurvivesTransformProperties() {
+ // transformProperties only retains keys in the JDBC allowlist; the
connect-timeout key must be
+ // on it, otherwise an operator-configured value is silently dropped and
the catalog always uses
+ // the 30000 ms default (the production-path bug behind the create-time
verification feature).
+ Map<String, String> createProperties =
+ ImmutableMap.<String, String>builder()
+ .put(JdbcConfig.JDBC_URL.getKey(), "jdbc:sqlite::memory:")
+ .put(JdbcConfig.JDBC_DRIVER.getKey(), "org.sqlite.JDBC")
+ .put(JdbcConfig.USERNAME.getKey(), "user")
+ .put(JdbcConfig.PASSWORD.getKey(), "password")
+ .put(JdbcConfig.CONNECT_TIMEOUT_MS.getKey(), "7500")
+ .build();
+
+ Map<String, String> transformed =
+ new
JdbcCatalogPropertiesMetadata().transformProperties(createProperties);
+
+ Assertions.assertEquals("7500",
transformed.get(JdbcConfig.CONNECT_TIMEOUT_MS.getKey()));
+ Assertions.assertEquals(
+ 7500L, new JdbcConfig(transformed).getConnectTimeoutMs(), "value must
reach JdbcConfig");
+ Assertions.assertFalse(
+ new JdbcCatalogPropertiesMetadata()
+ .isHiddenProperty(JdbcConfig.CONNECT_TIMEOUT_MS.getKey()));
+ }
+
/** Verifies the actual catalog configuration path for drivers without
isValid support. */
@ParameterizedTest
@CsvSource({
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestDataSourceValidation.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestDataSourceValidation.java
index 2636c8ba00..40db43de67 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestDataSourceValidation.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestDataSourceValidation.java
@@ -119,6 +119,32 @@ public class TestDataSourceValidation {
}
}
+ /**
+ * Verifies that for an unknown driver family (SQLite) no connect/socket
timeout connection
+ * property is injected and the datasource still builds and borrows a
connection. This guards the
+ * "unknown driver -> no-op" contract of {@link JdbcConnectionTimeouts}:
SQLite does not
+ * understand {@code connectTimeout}/{@code socketTimeout}, so injecting
them would make the
+ * connection fail.
+ */
+ @Test
+ public void testUnknownDriverInjectsNoConnectionTimeout() throws
SQLException {
+ // JdbcConnectionTimeouts emits nothing for the SQLite driver family.
+ Map<String, String> props = new HashMap<>();
+ props.put(JdbcConfig.JDBC_DRIVER.getKey(), "org.sqlite.JDBC");
+ props.put(JdbcConfig.JDBC_URL.getKey(), "jdbc:sqlite::memory:");
+ Assertions.assertTrue(
+ JdbcConnectionTimeouts.connectionProperties(
+ new JdbcConfig(props), props.get(JdbcConfig.JDBC_URL.getKey()))
+ .isEmpty());
+
+ // And the datasource built from it still borrows a working connection (no
leak, no failure).
+ try (BasicDataSource dataSource = createDataSource(Map.of())) {
+ try (Connection connection = dataSource.getConnection()) {
+ Assertions.assertTrue(connection.isValid(1));
+ }
+ }
+ }
+
/** A driver fixture whose connections do not implement JDBC connection
validation. */
public static class UnsupportedValidationDriver extends JDBC {
/**
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConfig.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConfig.java
index 9eebf08ff7..2b558ff22a 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConfig.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConfig.java
@@ -46,4 +46,47 @@ public class TestJdbcConfig {
jdbcConfig = new JdbcConfig(immutableMap);
Assertions.assertEquals(60000L, jdbcConfig.getMaxWaitMs());
}
+
+ @Test
+ void testConnectTimeoutMs() {
+ JdbcConfig jdbcConfig = new JdbcConfig(Maps.newHashMap());
+ Assertions.assertEquals(30000L, jdbcConfig.getConnectTimeoutMs());
+
+ ImmutableMap immutableMap =
ImmutableMap.of("jdbc.pool.connect-timeout-ms", "5000");
+ jdbcConfig = new JdbcConfig(immutableMap);
+ Assertions.assertEquals(5000L, jdbcConfig.getConnectTimeoutMs());
+ }
+
+ @Test
+ void testConnectTimeoutMsRejectsNonPositiveAndOverIntMax() {
+ // Config validation is lazy (checkValue runs on get()), so assert via the
getter.
+ // Zero / negative are rejected (must be positive).
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new JdbcConfig(ImmutableMap.of("jdbc.pool.connect-timeout-ms",
"0"))
+ .getConnectTimeoutMs());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new JdbcConfig(ImmutableMap.of("jdbc.pool.connect-timeout-ms",
"-1"))
+ .getConnectTimeoutMs());
+
+ // Integer.MAX_VALUE is the inclusive upper bound (the MySQL/PostgreSQL
driver timeouts are
+ // ints), so it is accepted and one more is rejected.
+ Assertions.assertEquals(
+ (long) Integer.MAX_VALUE,
+ new JdbcConfig(
+ ImmutableMap.of(
+ "jdbc.pool.connect-timeout-ms", Long.toString((long)
Integer.MAX_VALUE)))
+ .getConnectTimeoutMs());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new JdbcConfig(
+ ImmutableMap.of(
+ "jdbc.pool.connect-timeout-ms",
+ Long.toString((long) Integer.MAX_VALUE + 1L)))
+ .getConnectTimeoutMs());
+ }
}
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConnectionTimeouts.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConnectionTimeouts.java
new file mode 100644
index 0000000000..da3426ff26
--- /dev/null
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestJdbcConnectionTimeouts.java
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.catalog.jdbc.utils;
+
+import com.google.common.collect.Maps;
+import java.util.Map;
+import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJdbcConnectionTimeouts {
+
+ private static final String DRIVER_KEY = "jdbc-driver";
+ private static final String URL_KEY = "jdbc-url";
+ private static final String CONNECT_TIMEOUT_MS_KEY =
"jdbc.pool.connect-timeout-ms";
+ private static final String CONNECT_TIMEOUT = "connectTimeout";
+ private static final String SOCKET_TIMEOUT = "socketTimeout";
+ private static final String CLICKHOUSE_CONNECT_TIMEOUT = "connect_timeout";
+ private static final String CLICKHOUSE_SOCKET_TIMEOUT = "socket_timeout";
+
+ private static JdbcConfig config(Map<String, String> props) {
+ return new JdbcConfig(props);
+ }
+
+ @Test
+ void testMysqlFamilyEmitsConnectTimeoutMillisOnly() {
+ for (String driver : new String[] {"com.mysql.cj.jdbc.Driver",
"com.mysql.jdbc.Driver"}) {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, driver);
+ props.put(URL_KEY, "jdbc:mysql://host:3306/db");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ // Only the connect timeout is emitted (never a socket-read timeout), as
milliseconds.
+ Assertions.assertEquals("30000", result.get(CONNECT_TIMEOUT), driver);
+ Assertions.assertFalse(result.containsKey(SOCKET_TIMEOUT), driver);
+ Assertions.assertEquals(1, result.size(), driver);
+ }
+ }
+
+ @Test
+ void testOceanBaseDriverEmitsConnectTimeoutMillis() {
+ // OceanBase Connector/J is MySQL-compatible and uses the same camelCase
param in milliseconds.
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.oceanbase.jdbc.Driver");
+ props.put(URL_KEY, "jdbc:oceanbase://host:2881/db");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals("30000", result.get(CONNECT_TIMEOUT));
+ Assertions.assertFalse(result.containsKey(SOCKET_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testClickHouseDriverEmitsSnakeCaseConnectTimeoutMillis() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.clickhouse.jdbc.ClickHouseDriver");
+ props.put(URL_KEY, "jdbc:clickhouse://host:8443/default");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "7500");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ // ClickHouse uses the snake_case name and milliseconds; the camelCase
name must NOT be emitted,
+ // and no socket-read timeout is emitted.
+ Assertions.assertEquals("7500", result.get(CLICKHOUSE_CONNECT_TIMEOUT));
+ Assertions.assertFalse(result.containsKey(CLICKHOUSE_SOCKET_TIMEOUT));
+ Assertions.assertFalse(result.containsKey(CONNECT_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testClickHouseOperatorConnectTimeoutInJdbcUrlIsPreserved() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.clickhouse.jdbc.ClickHouseDriver");
+ String url = "jdbc:clickhouse://host:8443/default?connect_timeout=5000";
+ props.put(URL_KEY, url);
+
+ Map<String, String> result =
JdbcConnectionTimeouts.connectionProperties(config(props), url);
+
+ // The operator's connect_timeout wins, so nothing is injected.
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testHologresUsesPostgresqlDriverSeconds() {
+ // Hologres connects through the PostgreSQL driver, so it inherits the
seconds translation.
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.postgresql.Driver");
+ props.put(URL_KEY, "jdbc:postgresql://host:80/db");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "5000");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals("5", result.get(CONNECT_TIMEOUT));
+ Assertions.assertFalse(result.containsKey(SOCKET_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testPostgresRoundsUpToSecondsWithMinimumOne() {
+ assertPostgresConnectSeconds("30000", "30");
+ assertPostgresConnectSeconds("1", "1");
+ assertPostgresConnectSeconds("999", "1");
+ assertPostgresConnectSeconds("1000", "1");
+ assertPostgresConnectSeconds("1001", "2");
+ assertPostgresConnectSeconds("1999", "2");
+ }
+
+ @Test
+ void testPostgresSecondsConversionAtMaxAllowedValue() {
+ // The config caps the value at Integer.MAX_VALUE ms; the ceiling
conversion stays correct and
+ // within int range at that bound (and is overflow-safe for the whole
positive long range).
+ long millis = Integer.MAX_VALUE;
+ long expectedSeconds = (millis - 1L) / 1000L + 1L;
+ assertPostgresConnectSeconds(Long.toString(millis),
Long.toString(expectedSeconds));
+ Assertions.assertTrue(expectedSeconds > 1L, "conversion must not collapse
to 1 second");
+ Assertions.assertTrue(
+ expectedSeconds <= Integer.MAX_VALUE, "converted seconds must fit the
driver int range");
+ }
+
+ private static void assertPostgresConnectSeconds(String millis, String
expectedSeconds) {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.postgresql.Driver");
+ props.put(URL_KEY, "jdbc:postgresql://host:5432/db");
+ props.put(CONNECT_TIMEOUT_MS_KEY, millis);
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals(expectedSeconds, result.get(CONNECT_TIMEOUT),
millis);
+ Assertions.assertFalse(result.containsKey(SOCKET_TIMEOUT), millis);
+ }
+
+ @Test
+ void testUnknownDriverEmitsNothing() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.sqlite.JDBC");
+ props.put(URL_KEY, "jdbc:sqlite:/tmp/test.db");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testOperatorConnectTimeoutInConnectionPropertiesIsPreserved() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ props.put(URL_KEY, "jdbc:mysql://host:3306/db");
+ // The operator set connectTimeout inside the DBCP connectionProperties
value (name=value;...).
+ props.put("connectionProperties", "connectTimeout=5000;useSSL=false");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ // The operator's connectTimeout wins, so nothing is injected (no
overwrite).
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testBareTopLevelConnectTimeoutKeyDoesNotSuppressInjection() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ props.put(URL_KEY, "jdbc:mysql://host:3306/db");
+ // A bare top-level connectTimeout key is NOT a DBCP connection property
(DBCP ignores it), so
+ // it must NOT suppress injection — otherwise the connection would end up
with no timeout.
+ props.put(CONNECT_TIMEOUT, "5000");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals("30000", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testOperatorConnectTimeoutInJdbcUrlIsPreserved() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ String url = "jdbc:mysql://host:3306/db?connectTimeout=5000";
+ props.put(URL_KEY, url);
+
+ Map<String, String> result =
JdbcConnectionTimeouts.connectionProperties(config(props), url);
+
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testUrlParameterMatchIsCaseSensitiveAndBoundaryAnchored() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ // "ConnectTimeout" (wrong case) and "tunnelConnectTimeout" (substring)
must NOT suppress the
+ // canonical key, so the canonical connect timeout is still injected.
+ String url =
"jdbc:mysql://host:3306/db?ConnectTimeout=5000&tunnelConnectTimeout=6000";
+ props.put(URL_KEY, url);
+
+ Map<String, String> result =
JdbcConnectionTimeouts.connectionProperties(config(props), url);
+
+ Assertions.assertEquals("30000", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testCustomConnectTimeoutPropagates() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ props.put(URL_KEY, "jdbc:mysql://host:3306/db");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "7500");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.connectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals("7500", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testNullJdbcUrlIsSafe() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.postgresql.Driver");
+
+ Map<String, String> result =
JdbcConnectionTimeouts.connectionProperties(config(props), null);
+
+ Assertions.assertEquals("30", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals(1, result.size());
+ }
+
+ @Test
+ void testMysqlProbeEmitsConnectAndSocketTimeoutMillis() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ props.put(URL_KEY, "jdbc:mysql://host:3306/db");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "2000");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.probeConnectionProperties(config(props),
props.get(URL_KEY));
+
+ // The probe bounds both the connect and the read timeout (milliseconds
for MySQL).
+ Assertions.assertEquals("2000", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals("2000", result.get(SOCKET_TIMEOUT));
+ Assertions.assertEquals(2, result.size());
+ }
+
+ @Test
+ void testPostgresProbeEmitsConnectSocketAndLoginTimeoutSeconds() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.postgresql.Driver");
+ props.put(URL_KEY, "jdbc:postgresql://host:5432/db");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "2000");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.probeConnectionProperties(config(props),
props.get(URL_KEY));
+
+ // The probe bounds connect, socket and login timeouts, converted to
seconds.
+ Assertions.assertEquals("2", result.get(CONNECT_TIMEOUT));
+ Assertions.assertEquals("2", result.get(SOCKET_TIMEOUT));
+ Assertions.assertEquals("2", result.get("loginTimeout"));
+ Assertions.assertEquals(3, result.size());
+ }
+
+ @Test
+ void testClickHouseProbeEmitsSnakeCaseConnectAndSocketTimeoutMillis() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.clickhouse.jdbc.ClickHouseDriver");
+ props.put(URL_KEY, "jdbc:clickhouse://host:8443/default");
+ props.put(CONNECT_TIMEOUT_MS_KEY, "2000");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.probeConnectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertEquals("2000", result.get(CLICKHOUSE_CONNECT_TIMEOUT));
+ Assertions.assertEquals("2000", result.get(CLICKHOUSE_SOCKET_TIMEOUT));
+ Assertions.assertFalse(result.containsKey(CONNECT_TIMEOUT));
+ Assertions.assertEquals(2, result.size());
+ }
+
+ @Test
+ void testProbeUnknownDriverEmitsNothing() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "org.sqlite.JDBC");
+ props.put(URL_KEY, "jdbc:sqlite::memory:");
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.probeConnectionProperties(config(props),
props.get(URL_KEY));
+
+ Assertions.assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testProbeOperatorSocketTimeoutIsPreserved() {
+ Map<String, String> props = Maps.newHashMap();
+ props.put(DRIVER_KEY, "com.mysql.cj.jdbc.Driver");
+ String url = "jdbc:mysql://host:3306/db?socketTimeout=9000";
+ props.put(URL_KEY, url);
+
+ Map<String, String> result =
+ JdbcConnectionTimeouts.probeConnectionProperties(config(props), url);
+
+ // The operator's socketTimeout wins; only the connect timeout is injected.
+ Assertions.assertFalse(result.containsKey(SOCKET_TIMEOUT));
+ Assertions.assertEquals("30000", result.get(CONNECT_TIMEOUT));
+ }
+}
diff --git
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
index 6fef5372f9..8b920576e1 100644
---
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
@@ -216,6 +216,10 @@ public class CatalogMysqlIT extends BaseIT {
JdbcConfig.JDBC_DRIVER.getKey(),
MYSQL_CONTAINER.getDriverClassName(TEST_DB_NAME));
catalogProperties.put(JdbcConfig.USERNAME.getKey(),
MYSQL_CONTAINER.getUsername());
catalogProperties.put(JdbcConfig.PASSWORD.getKey(),
MYSQL_CONTAINER.getPassword());
+ // This test registers a catalog pointing at an unreachable endpoint on
purpose, so it can then
+ // assert testConnection fails on the EXISTING catalog. It must opt out of
create-time
+ // verification; the rejection by the default create-time verification is
covered separately.
+ catalogProperties.put(JdbcConfig.VERIFY_ON_CREATE.getKey(), "false");
metalake.createCatalog(
unreachableCatalogName,
@@ -2433,6 +2437,10 @@ public class CatalogMysqlIT extends BaseIT {
String password = MYSQL_CONTAINER.getPassword();
String wrongPassword = password + "wrong";
catalogProperties.put(JdbcConfig.PASSWORD.getKey(), wrongPassword);
+ // This test deliberately creates the catalog with a wrong password and
then fixes it via
+ // alterCatalog, so it must opt out of create-time backend verification;
otherwise the create
+ // itself would be rejected for the unreachable backend (the behavior
added in #13767).
+ catalogProperties.put(JdbcConfig.VERIFY_ON_CREATE.getKey(), "false");
metalake.createCatalog(
testCatalogName, Catalog.Type.RELATIONAL, provider, "comment",
catalogProperties);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 2a23d7e0fa..16b5d6a9e1 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -996,7 +996,11 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
.build())
.build();
- CatalogWrapper wrapper = createCatalogWrapper(dummyEntity, mergedConfig);
+ // Validate the properties (reserved/required/hidden-mask checks) but do
NOT run the
+ // create-time backend probe here: this standalone path performs its own
explicit probe
+ // below, and letting the wrapper validate as well would probe the
backend twice.
+ CatalogWrapper wrapper =
+ createCatalogWrapper(dummyEntity, mergedConfig, false /*
validateBackend */);
try {
wrapper.doWithCatalogOps(
c -> {
@@ -1772,11 +1776,33 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
*/
CatalogWrapper createCatalogWrapper(
CatalogEntity entity, @Nullable Map<String, String> propsToValidate) {
+ // Default: when properties are validated at create, also verify the
backend. The standalone
+ // testConnection path uses the overload below to keep property validation
but skip the probe.
+ return createCatalogWrapper(entity, propsToValidate, true /*
validateBackend */);
+ }
+
+ /**
+ * Create a catalog wrapper, optionally running the create-time backend
probe. {@code
+ * propsToValidate} always drives property validation when non-null; {@code
validateBackend}
+ * additionally controls whether {@link BaseCatalog#validateOnCreate()}
runs. The standalone
+ * {@code testConnection} path passes {@code validateBackend = false} so it
still rejects invalid
+ * properties while performing its own single explicit probe.
+ *
+ * @param entity The catalog entity.
+ * @param propsToValidate The properties to validate; may be null to skip
property validation.
+ * @param validateBackend Whether to run the create-time backend probe.
+ * @return The created catalog wrapper.
+ */
+ CatalogWrapper createCatalogWrapper(
+ CatalogEntity entity,
+ @Nullable Map<String, String> propsToValidate,
+ boolean validateBackend) {
Map<String, String> conf = entity.getProperties();
String provider = entity.getProvider();
if (!classLoaderSharingEnabled) {
- return createNonPooledCatalogWrapper(provider, conf, entity,
propsToValidate);
+ return createNonPooledCatalogWrapper(
+ provider, conf, entity, propsToValidate, validateBackend);
}
ClassLoaderKey key = buildClassLoaderKey(provider, conf);
@@ -1787,7 +1813,8 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
initCatalogWrapper(
new CatalogWrapper(poolEntry.classLoader(), classLoaderPool,
poolEntry),
entity,
- propsToValidate);
+ propsToValidate,
+ validateBackend);
return wrapper;
} catch (Exception e) {
classLoaderPool.release(poolEntry);
@@ -1799,10 +1826,12 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
String provider,
Map<String, String> conf,
CatalogEntity entity,
- @Nullable Map<String, String> propsToValidate) {
+ @Nullable Map<String, String> propsToValidate,
+ boolean validateBackend) {
IsolatedClassLoader classLoader = createClassLoader(provider, conf);
try {
- return initCatalogWrapper(new CatalogWrapper(classLoader), entity,
propsToValidate);
+ return initCatalogWrapper(
+ new CatalogWrapper(classLoader), entity, propsToValidate,
validateBackend);
} catch (Exception e) {
ClassLoaderPool.cleanupClassLoader(classLoader);
throw e;
@@ -1814,7 +1843,10 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
* into the given wrapper.
*/
private CatalogWrapper initCatalogWrapper(
- CatalogWrapper wrapper, CatalogEntity entity, @Nullable Map<String,
String> propsToValidate) {
+ CatalogWrapper wrapper,
+ CatalogEntity entity,
+ @Nullable Map<String, String> propsToValidate,
+ boolean validateBackend) {
BaseCatalog<?> catalog = createBaseCatalog(wrapper.classLoader, entity);
wrapper.catalog = catalog;
try {
@@ -1826,11 +1858,14 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
catalog.properties();
catalog.capability();
- // Eagerly initialize opted-in catalogs so a misconfiguration
fails at create rather
- // than on first use. The backend translates failures into the
caller-error vs
+ // Validate opted-in catalogs so a misconfiguration fails at
create rather than on
+ // first use. The default validation eagerly initializes ops();
JDBC overrides it to
+ // run a real backend probe. The backend translates failures into
the caller-error vs
// dependency-unavailable taxonomy; both types are forwarded by
the passthrough below.
- if (propsToValidate != null && catalog.shouldValidateOnCreate()) {
- catalog.ops();
+ // The backend probe is gated by validateBackend so the standalone
testConnection path
+ // can keep property validation while running its own explicit
probe instead.
+ if (validateBackend && propsToValidate != null &&
catalog.shouldValidateOnCreate()) {
+ catalog.validateOnCreate();
}
return null;
},
diff --git a/core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java
b/core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java
index 801fd3bf45..4b36da86e1 100644
--- a/core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java
+++ b/core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java
@@ -227,6 +227,21 @@ public abstract class BaseCatalog<T extends BaseCatalog>
return false;
}
+ /**
+ * Validates this catalog at create time when {@link
#shouldValidateOnCreate()} returns true. The
+ * default eagerly initializes {@link #ops()}, which preserves the current
behavior for every
+ * existing opt-in (for example Iceberg REST, which only needs eager
initialization). Subclasses
+ * that can cheaply probe their backend — such as JDBC — override this to
run a real connection
+ * check and throw a connection-failed error when the backend is unreachable.
+ *
+ * <p>This is {@code public} rather than {@code protected} because {@code
CatalogManager}, in a
+ * different package, invokes it on the create-time validation path
(mirroring {@link #ops()} and
+ * {@link #shouldValidateOnCreate()}).
+ */
+ public void validateOnCreate() {
+ ops();
+ }
+
public void checkMetalakeAndCatalogInUse() {
Map<String, String> catalogProperties = entity.getProperties();
if (catalogProperties == null) {
diff --git a/core/src/test/java/org/apache/gravitino/TestCatalog.java
b/core/src/test/java/org/apache/gravitino/TestCatalog.java
index 4e8e8ada24..d3fb322182 100644
--- a/core/src/test/java/org/apache/gravitino/TestCatalog.java
+++ b/core/src/test/java/org/apache/gravitino/TestCatalog.java
@@ -67,6 +67,23 @@ public class TestCatalog extends BaseCatalog<TestCatalog> {
&&
Boolean.parseBoolean(properties.get(TestCatalogOperations.VALIDATE_ON_CREATE));
}
+ /**
+ * Mirrors the JDBC override: runs a real {@code testConnection} probe at
create time instead of
+ * only eagerly initializing {@code ops()}. {@code ops()} is still invoked
(so a failing {@code
+ * initialize()} still surfaces here), and the probe additionally throws
when the catalog was
+ * created with {@code need-fail=true}, exercising the create-time
verification/rollback path.
+ */
+ @Override
+ public void validateOnCreate() {
+ try {
+ ops().testConnection(entity().nameIdentifier());
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
protected Capability newCapability() {
return new TestCatalogCapabilities();
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 90dff3e5e0..6dbc3662f0 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -26,6 +26,7 @@ import static org.apache.gravitino.TestCatalog.PROPERTY_KEY3;
import static org.apache.gravitino.TestCatalog.PROPERTY_KEY4;
import static org.apache.gravitino.TestCatalog.PROPERTY_KEY5_PREFIX;
import static org.apache.gravitino.TestCatalog.PROPERTY_KEY6_PREFIX;
+import static org.apache.gravitino.TestCatalog.PROPERTY_RESERVED_KEY;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
@@ -70,6 +71,7 @@ import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
import org.apache.gravitino.exceptions.CatalogNotInUseException;
+import org.apache.gravitino.exceptions.ConnectionFailedException;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -625,6 +627,110 @@ public class TestCatalogManager {
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(skipIdent));
}
+ @Test
+ public void testCreateCatalogVerifiesBackendViaTestConnection() throws
Exception {
+ // Opted in via validateOnCreate() force-probe, and the probe reports the
backend unreachable:
+ // creation must propagate ConnectionFailedException and roll back both
the cache and the store.
+ NameIdentifier failIdent = NameIdentifier.of("metalake",
"verify-testconn-fail");
+ Map<String, String> failProps =
+ Maps.newHashMap(
+ ImmutableMap.of(
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3",
+ TestCatalogOperations.VALIDATE_ON_CREATE,
+ "true",
+ TestCatalogOperations.FAIL_TEST,
+ "true"));
+
+ Assertions.assertThrows(
+ ConnectionFailedException.class,
+ () ->
+ catalogManager.createCatalog(
+ failIdent, Catalog.Type.RELATIONAL, provider, "comment",
failProps));
+
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(failIdent));
+ // Store rollback: the just-persisted entity is deleted.
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () -> entityStore.get(failIdent, EntityType.CATALOG,
CatalogEntity.class));
+
+ // Rolled back, so the identifier is reusable with a reachable backend.
+ Map<String, String> okProps =
+ Maps.newHashMap(
+ ImmutableMap.of(
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3",
+ TestCatalogOperations.VALIDATE_ON_CREATE,
+ "true"));
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager.createCatalog(
+ failIdent, Catalog.Type.RELATIONAL, provider, "comment",
okProps));
+
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(failIdent));
+
+ // Not opted in (no VALIDATE_ON_CREATE): the backend probe is never
forced, so a FAIL_TEST
+ // catalog is still created. This mirrors model/generic/non-JDBC catalogs
that are never
+ // force-verified.
+ NameIdentifier noOptInIdent = NameIdentifier.of("metalake",
"verify-no-optin");
+ Map<String, String> noOptInProps =
+ Maps.newHashMap(
+ ImmutableMap.of(
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3",
+ TestCatalogOperations.FAIL_TEST,
+ "true"));
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalogManager.createCatalog(
+ noOptInIdent, Catalog.Type.RELATIONAL, provider, "comment",
noOptInProps));
+
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(noOptInIdent));
+ }
+
+ @Test
+ public void testTestConnectionStillValidatesProperties() {
+ // Regression: the standalone testConnection path skips only the backend
probe, NOT property
+ // validation. A reserved property, or a missing required property, must
be rejected exactly as
+ // createCatalog would reject it (otherwise testConnection could report
success for a
+ // configuration createCatalog then refuses).
+ NameIdentifier ident = NameIdentifier.of("metalake",
"testconn-prop-validation");
+
+ Map<String, String> reservedProps = Maps.newHashMap();
+ reservedProps.put(PROPERTY_KEY1, "value1");
+ reservedProps.put(PROPERTY_KEY2, "value2");
+ reservedProps.put(PROPERTY_KEY5_PREFIX + "1", "value3");
+ reservedProps.put(PROPERTY_RESERVED_KEY, "value");
+ IllegalArgumentException reserved =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalogManager.testConnection(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
reservedProps));
+ Assertions.assertTrue(reserved.getMessage().contains("reserved"),
reserved.getMessage());
+
+ Map<String, String> missingRequiredProps = Maps.newHashMap();
+ // key1 is required; omit it.
+ missingRequiredProps.put(PROPERTY_KEY2, "value2");
+ missingRequiredProps.put(PROPERTY_KEY5_PREFIX + "1", "value3");
+ IllegalArgumentException missing =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalogManager.testConnection(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
missingRequiredProps));
+ Assertions.assertTrue(missing.getMessage().contains("required"),
missing.getMessage());
+ }
+
@Test
public void testListCatalogs() {
NameIdentifier ident = NameIdentifier.of("metalake", "test11");
diff --git
a/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
b/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
index 5fd405c62f..3ac659243c 100644
---
a/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
+++
b/core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java
@@ -130,6 +130,8 @@ public class TestCatalogOperations
public final Map<NameIdentifier, View> views;
+ private final Map<String, String> config;
+
public static final String FAIL_CREATE = "fail-create";
public static final String FAIL_TEST = "need-fail";
@@ -149,6 +151,7 @@ public class TestCatalogOperations
modelVersions = Maps.newHashMap();
modelAliasToVersion = Maps.newHashMap();
views = Maps.newHashMap();
+ this.config = config == null ? ImmutableMap.of() :
ImmutableMap.copyOf(config);
}
@Override
@@ -886,6 +889,22 @@ public class TestCatalogOperations
}
}
+ @Override
+ public void testConnection(NameIdentifier catalogIdent) {
+ // Mirror a JDBC-style backend probe: fail when the catalog was created
with FAIL_TEST=true.
+ if ("true".equals(config.get(FAIL_TEST))) {
+ throw new ConnectionFailedException("Connection failed");
+ }
+ // A catalog that opts into create-time verification mirrors a reachable
JDBC backend: the probe
+ // succeeds. Otherwise keep the interface default (unsupported) so the
standalone testConnection
+ // REST path still reports that this test catalog does not support
connection testing.
+ if ("true".equals(config.get(VALIDATE_ON_CREATE))) {
+ return;
+ }
+ throw new UnsupportedOperationException(
+ String.format("Catalog %s does not support connection testing",
catalogIdent));
+ }
+
@Override
public void testConnection(
NameIdentifier name,
diff --git a/docs/jdbc-catalog-connection-validation.md
b/docs/jdbc-catalog-connection-validation.md
index b313982491..13cedaafd7 100644
--- a/docs/jdbc-catalog-connection-validation.md
+++ b/docs/jdbc-catalog-connection-validation.md
@@ -55,3 +55,53 @@ a connection when initializing its connection factory.
For MySQL drivers affected by the catalog-switching problem, explicitly
setting `SELECT 1` restores
the cached-statement validation path and can reproduce connection churn. Leave
the query unset when
the driver supports `Connection.isValid()`.
+
+## Backend verification at create time
+
+A relational JDBC catalog verifies that its backend is reachable when the
catalog is created,
+rather than deferring the first connection to a later operation. The
verification opens a dedicated,
+time-bounded one-off probe connection (separate from the pool used by normal
operations), confirms
+it, and closes it. If it fails, the create is rejected and rolled back (the
catalog entity is not
+persisted), and the caller receives a connection error that names the backend
`host:port` and the
+underlying driver message.
+
+To register a catalog whose backend is temporarily unavailable, set the hidden
catalog property
+`verify-on-create` to `false`, which skips the create-time check:
+
+```json
+{
+ "verify-on-create": "false"
+}
+```
+
+The default is `true` (verify on create). The connection is still opened
lazily on first use when
+the check is skipped.
+
+### Bounded timeouts
+
+So that create-time verification fails fast instead of hanging, the timeout
+`jdbc.pool.connect-timeout-ms` (default `30000` ms, maximum `2147483647`)
bounds a JDBC connection.
+Gravitino translates this canonical millisecond value into each driver
family's own parameters and
+unit:
+
+- MySQL Connector/J — also used by the Doris, StarRocks, and OceanBase
catalogs: `connectTimeout`
+ and `socketTimeout` in milliseconds.
+- PostgreSQL — also used by the Hologres catalog: `connectTimeout`,
`socketTimeout` and
+ `loginTimeout` in seconds (rounded up, minimum 1).
+- ClickHouse: `connect_timeout` and `socket_timeout` in milliseconds (note the
underscore names).
+
+It is applied differently to the two kinds of connection:
+
+- **Pooled connections** (normal operations) get only the **connect** timeout.
A socket-read
+ timeout is deliberately not imposed there, because it would also abort
legitimate long-running
+ operations on a healthy backend (a large `ALTER TABLE`, a slow metadata
query).
+- **The create-time verification probe** gets both the **connect** and the
**read** (and, for
+ PostgreSQL, **login**) timeout. The probe runs under the metalake write
lock, so it must be fully
+ bounded: a backend that accepts the TCP connection but never speaks the
database protocol (a load
+ balancer with a dead backend, a wrong port, a proxy that accepts and never
forwards) would
+ otherwise block a connect-only bound forever and stall the whole metalake.
+
+A timeout value you set directly on the connection or in the `jdbc-url` query
string is never
+overridden; drivers whose family is not recognized are left unbounded by
Gravitino. The pooled
+timeout is applied as a per-connection bound, so the pool stays lazy and no
connection is opened
+until the first real round trip.
