jerryshao commented on code in PR #14508:
URL: https://github.com/apache/gravitino/pull/14508#discussion_r4464289703
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java:
##########
@@ -485,31 +496,72 @@ private List<SchemaPO> listSchemaPOs(Namespace namespace)
{
mapper -> POStorageReadRouting.listPOs(mapper, namespace, ops,
Entity.EntityType.SCHEMA));
}
+ private boolean prepareSchemaOverwrite(
+ NameIdentifier identifier, SchemaPO incoming, SchemaMetaMapper mapper) {
+ // Reject an active foreign identity before taking its row lock. The
locking read below
+ // must still revalidate identity because the first check is only a
fail-fast optimization.
+ SchemaPO registeredId =
mapper.selectSchemaMetaById(incoming.getSchemaId());
+ rejectForeignSchemaIdentity(identifier, incoming, registeredId);
+ // Lock even a retired ID before reviving it. Otherwise two catalogs could
both revive the
+ // same marker and the slower upsert would move the winner's newly live
registration.
+ registeredId =
mapper.selectSchemaMetaByIdIncludingDeletedForUpdate(incoming.getSchemaId());
+ rejectForeignSchemaIdentity(identifier, incoming, registeredId);
+ SchemaPO stored =
+ ops.listPOs(mapper, identifier.namespace(),
List.of(identifier.name())).stream()
+ .findFirst()
+ .orElse(null);
+ if (stored != null &&
!stored.getSchemaId().equals(incoming.getSchemaId())) {
Review Comment:
**Major: an overwrite can cascade-delete metadata for a schema that is still
live in the source.**
When the stored row for this name has a different id,
`prepareSchemaOverwrite` treats it as stale and cascade-deletes it
(`deleteSchemaAndGet` at line 523). It never checks whether that stored
identity still exists in the source. `importSchema` does make this check
(`SchemaOperationDispatcher.java:687`) and refuses to move a live schema, but
`createSchema` always calls `store.put(..., true)`
(`SchemaOperationDispatcher.java:247`), so a new id is the normal case there.
Scenario: the schema is renamed externally from `s` to `s2`, and nobody has
loaded `s2` yet. `createSchema(s)` succeeds in the source, then drops `s`'s
tags, privileges and child registrations, even though that schema is still live
as `s2`.
Suggestion: before the cascade, resolve the stored id in the source the same
way `importSchema` does, and skip the delete (or fail) if that object still
exists.
##########
core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java:
##########
@@ -229,12 +247,11 @@ public Schema createSchema(
store.put(schemaEntity, true /* overwrite */);
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", ident,
e);
- return EntityCombinedSchema.of(schema)
- .withHiddenProperties(
- getMaskAndOmitKeys(
- catalogIdent,
- HasPropertyMetadata::schemaPropertiesMetadata,
- schema.properties()));
+ throw new GravitinoRuntimeException(
Review Comment:
**Major: there's no real recovery path after the source create succeeds but
registration fails.**
`GravitinoRuntimeException` maps to HTTP 500
(`ExceptionHandlers.java:1283`). The message suggests retrying, but a retry
hits `SchemaAlreadyExistsException` and registers nothing. The external schema
and its secrets (line 268) stay unregistered until a later load happens to
import them. The same applies to tables at `TableOperationDispatcher.java:802`.
Suggestion: either make the message say that recovery is a load/import, or,
on already-exists after this kind of failure, register the existing source
object instead of failing again. A test that does fail-then-retry would help,
since `TestSchemaOperationDispatcher` currently only checks that the first
failure keeps the secret.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java:
##########
@@ -175,10 +180,10 @@ public void insertSchema(SchemaEntity schemaEntity,
boolean overwrite) throws IO
// Everything below runs in one transaction, and it starts by locking
the parent catalog row.
// That lock is what stops a catalog drop from running at the same time
as this insert. A
- // plain name is enough with a shared lock; a nested name needs an
exclusive one, see
- // lockCatalogForSchemaCreate.
+ // plain insert can use a shared lock; nested ancestors or an overwrite
need an exclusive
+ // lock because replacement can retire a stale schema and its
descendants.
SessionUtils.doMultipleWithCommit(
- () -> lockCatalogForSchemaCreate(catalogPO, rowsToInsert.size() > 1),
+ () -> lockCatalogForSchemaCreate(catalogPO, rowsToInsert.size() > 1
|| overwrite),
Review Comment:
Minor: every overwrite takes the exclusive parent-catalog lock, and external
`createSchema` always overwrites. So even creating a new name with no stale row
serializes every schema create and drop in the catalog. Could we keep the
shared lock when the name lookup finds no row with a different id, and only
roll back and rerun the transaction under the exclusive lock when it does?
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -165,18 +165,37 @@ public NameIdentifier[] listTables(Namespace namespace)
throws NoSuchSchemaExcep
@Override
public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
EntityCombinedTable entityCombinedTable =
- TreeLockUtils.doWithTreeLock(ident, LockType.READ, () ->
internalLoadTable(ident));
-
- if (!entityCombinedTable.imported()) {
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
+ () -> {
+ // A cold external table will be loaded under its own WRITE lock
below. Do not
+ // fetch it here only to discard the snapshot and fetch it again
during import.
+ // Managed catalogs do not need the external import path, even
without a store row.
+ if (getEntity(ident, TABLE, TableEntity.class) == null
+ && !isManagedEntity(getCatalogIdentifier(ident),
Capability.Scope.TABLE)) {
+ return null;
+ }
+ return internalLoadTable(ident);
+ });
+
+ if (entityCombinedTable == null || !entityCombinedTable.imported()) {
// Load the schema to make sure the schema is imported.
SchemaDispatcher schemaDispatcher = getSchemaDispatcher();
NameIdentifier schemaIdent =
NameIdentifier.of(ident.namespace().levels());
- schemaDispatcher.loadSchema(schemaIdent);
+ try {
+ schemaDispatcher.loadSchema(schemaIdent);
+ } catch (NoSuchSchemaException e) {
Review Comment:
Minor: on table load, a missing schema used to surface as
`NoSuchSchemaException`; with this change it comes back as
`NoSuchTableException`. REST uses the outer exception class name
(`server-common/.../web/Utils.java:163`), so clients lose the schema-missing
type, while #14499 asks to preserve missing-schema errors. Could we propagate
`NoSuchSchemaException` here?
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -778,7 +799,11 @@ private Table internalCreateTable(
store.put(tableEntity, true /* overwrite */);
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", ident, e);
- return
EntityCombinedTable.of(table).withHiddenProperties(catalogResult.hiddenProperties);
+ throw new GravitinoRuntimeException(
Review Comment:
Same issue as the schema case in `SchemaOperationDispatcher`: retrying after
this failure returns `TableAlreadyExistsException`, and the table stays
unregistered until a later load imports it.
##########
core/src/main/java/org/apache/gravitino/utils/BoundedConnectionDriver.java:
##########
@@ -0,0 +1,295 @@
+/*
+ * 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.utils;
+
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.security.ProtectionDomain;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Bounds physical connection creation for drivers that cannot enforce a login
timeout.
+ *
+ * <p>Only {@link Driver#connect} runs on a worker. Validation, metadata
queries and DDL stay on the
+ * caller. A connection that arrives after timeout or interruption is closed
by its worker and is
+ * never handed to a pool. Cancellation is best effort: a driver ignoring
interruption retains a
+ * worker until it returns. A JVM-wide limit of 64 workers, with no queue,
bounds that retention;
+ * excess attempts fail with {@link SQLTransientConnectionException}. Workers
are daemon threads and
+ * release their catalog context class loader after each attempt. Each
attempt, including late
+ * connection cleanup, runs with the submitting caller's access-control
context and JAAS subject;
+ * shared workers do not inherit the identity of the request that creates them.
+ *
+ * <p>After a timeout, further attempts through this instance fail immediately
until the abandoned
+ * attempt finishes cleanup. This prevents a lazy pool's initialization lock
from serializing a
+ * backlog of new attempts against the same unreachable backend. Normal pool
waits and SQL execution
+ * have their own budgets and are not bounded by this adapter.
+ */
+@SuppressWarnings("removal") // JDK 17 associates JAAS subjects with
access-control contexts.
+public final class BoundedConnectionDriver implements Driver {
+ private static final Logger LOG =
LoggerFactory.getLogger(BoundedConnectionDriver.class);
+
+ private static final AccessControlContext WORKER_CONTEXT =
+ new AccessControlContext(new ProtectionDomain[0]);
+
+ private static final ExecutorService WORKERS =
+ new ThreadPoolExecutor(
+ 0,
+ 64,
+ 60,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ task ->
+ AccessController.doPrivileged(
+ (PrivilegedAction<Thread>)
+ () -> {
+ Thread thread = new Thread(task,
"jdbc-connect-timeout");
+ thread.setDaemon(true);
+ thread.setContextClassLoader(
+ BoundedConnectionDriver.class.getClassLoader());
+ return thread;
+ },
+ WORKER_CONTEXT));
+
+ private final Driver delegate;
+ private final long timeoutMs;
+ private final ExecutorService workers;
+ private final AtomicInteger abandonedAttempts = new AtomicInteger();
+
+ /**
+ * Wraps a driver with a physical connection deadline.
+ *
+ * @param delegate the driver to call, without registering the wrapper with
DriverManager
+ * @param timeoutMs a positive timeout in milliseconds
+ * @throws IllegalArgumentException if timeoutMs is not positive
+ */
+ public BoundedConnectionDriver(Driver delegate, long timeoutMs) {
+ this(delegate, timeoutMs, WORKERS);
+ }
+
+ BoundedConnectionDriver(Driver delegate, long timeoutMs, ExecutorService
workers) {
+ if (timeoutMs <= 0) {
+ throw new IllegalArgumentException("JDBC connect timeout must be
positive");
+ }
+ this.delegate = delegate;
+ this.timeoutMs = timeoutMs;
+ this.workers = workers;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @throws SQLTimeoutException if the connection deadline expires or the
caller is interrupted
+ * @throws SQLTransientConnectionException if all workers are busy or a
previous timed-out attempt
+ * through this driver is still running
+ */
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException {
+ if (abandonedAttempts.get() != 0) {
+ throw new SQLTransientConnectionException(
+ "A previous timed-out JDBC connection attempt is still running. Try
again after it finishes.",
+ "08001");
+ }
+ Attempt attempt = new Attempt();
+ ClassLoader context = Thread.currentThread().getContextClassLoader();
+ AccessControlContext callerContext = AccessController.getContext();
+ Properties properties = (Properties) info.clone();
+ try {
+ workers.execute(
+ () ->
+ AccessController.doPrivileged(
+ (PrivilegedAction<Void>)
+ () -> {
+ attempt.run(url, properties, context);
+ return null;
+ },
+ callerContext));
+ } catch (RejectedExecutionException e) {
+ throw new SQLTransientConnectionException(
+ "The JDBC connection attempt limit has been reached. Try again after
an attempt finishes.",
+ "08001",
+ e);
+ }
+ try {
+ Connection connection = attempt.result.get(timeoutMs,
TimeUnit.MILLISECONDS);
+ attempt.accept();
+ return connection;
+ } catch (TimeoutException e) {
+ attempt.abandon();
+ throw new SQLTimeoutException(
+ "JDBC connection attempt exceeded "
+ + timeoutMs
+ + " ms. Check backend and proxy connectivity.",
+ "08001",
+ e);
+ } catch (InterruptedException e) {
+ attempt.abandon();
+ Thread.currentThread().interrupt();
+ throw new SQLTimeoutException("Interrupted while opening a JDBC
connection", "08001", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof SQLException) {
+ throw (SQLException) cause;
+ }
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new SQLException("Failed to open JDBC connection", "08001", cause);
+ } finally {
+ attempt.decision.countDown();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean acceptsURL(String url) throws SQLException {
+ return delegate.acceptsURL(url);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info)
throws SQLException {
+ return delegate.getPropertyInfo(url, info);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int getMajorVersion() {
+ return delegate.getMajorVersion();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int getMinorVersion() {
+ return delegate.getMinorVersion();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean jdbcCompliant() {
+ return delegate.jdbcCompliant();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public java.util.logging.Logger getParentLogger() throws
SQLFeatureNotSupportedException {
+ return delegate.getParentLogger();
+ }
+
+ private final class Attempt {
+ private final CompletableFuture<Connection> result = new
CompletableFuture<>();
+ private final CountDownLatch decision = new CountDownLatch(1);
+ private boolean accepted;
+ private boolean abandoned;
+ private boolean finished;
+ @Nullable private Thread runner;
+
+ synchronized void accept() {
+ accepted = true;
+ }
+
+ synchronized void abandon() {
+ if (!finished) {
+ abandoned = true;
+ abandonedAttempts.incrementAndGet();
+ if (runner != null) {
+ runner.interrupt();
+ }
+ }
+ }
+
+ void run(String url, Properties properties, @Nullable ClassLoader context)
{
+ Thread current = Thread.currentThread();
+ ClassLoader previous = current.getContextClassLoader();
+ current.setContextClassLoader(context);
+ Connection connection = null;
+ try {
+ synchronized (this) {
+ runner = current;
+ if (abandoned) {
+ return;
+ }
+ }
+ connection = delegate.connect(url, properties);
+ result.complete(connection);
+ boolean interrupted = false;
+ while (true) {
+ try {
+ decision.await();
+ break;
+ } catch (InterruptedException e) {
+ // Cancellation must not bypass the handoff decision or discard a
late connection.
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ current.interrupt();
Review Comment:
nit: re-interrupting the pooled worker isn't needed, because
`ThreadPoolExecutor` clears the flag in `getTask`, so the next task isn't
interrupted anyway. You can drop this; the pool already ends the task.
##########
core/src/main/java/org/apache/gravitino/utils/BoundedConnectionDriver.java:
##########
@@ -0,0 +1,295 @@
+/*
+ * 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.utils;
+
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.security.ProtectionDomain;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Bounds physical connection creation for drivers that cannot enforce a login
timeout.
+ *
+ * <p>Only {@link Driver#connect} runs on a worker. Validation, metadata
queries and DDL stay on the
+ * caller. A connection that arrives after timeout or interruption is closed
by its worker and is
+ * never handed to a pool. Cancellation is best effort: a driver ignoring
interruption retains a
+ * worker until it returns. A JVM-wide limit of 64 workers, with no queue,
bounds that retention;
+ * excess attempts fail with {@link SQLTransientConnectionException}. Workers
are daemon threads and
+ * release their catalog context class loader after each attempt. Each
attempt, including late
+ * connection cleanup, runs with the submitting caller's access-control
context and JAAS subject;
+ * shared workers do not inherit the identity of the request that creates them.
+ *
+ * <p>After a timeout, further attempts through this instance fail immediately
until the abandoned
+ * attempt finishes cleanup. This prevents a lazy pool's initialization lock
from serializing a
+ * backlog of new attempts against the same unreachable backend. Normal pool
waits and SQL execution
+ * have their own budgets and are not bounded by this adapter.
+ */
+@SuppressWarnings("removal") // JDK 17 associates JAAS subjects with
access-control contexts.
+public final class BoundedConnectionDriver implements Driver {
+ private static final Logger LOG =
LoggerFactory.getLogger(BoundedConnectionDriver.class);
+
+ private static final AccessControlContext WORKER_CONTEXT =
+ new AccessControlContext(new ProtectionDomain[0]);
+
+ private static final ExecutorService WORKERS =
+ new ThreadPoolExecutor(
+ 0,
+ 64,
+ 60,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ task ->
+ AccessController.doPrivileged(
+ (PrivilegedAction<Thread>)
+ () -> {
+ Thread thread = new Thread(task,
"jdbc-connect-timeout");
+ thread.setDaemon(true);
+ thread.setContextClassLoader(
+ BoundedConnectionDriver.class.getClassLoader());
+ return thread;
+ },
+ WORKER_CONTEXT));
+
+ private final Driver delegate;
+ private final long timeoutMs;
+ private final ExecutorService workers;
+ private final AtomicInteger abandonedAttempts = new AtomicInteger();
+
+ /**
+ * Wraps a driver with a physical connection deadline.
+ *
+ * @param delegate the driver to call, without registering the wrapper with
DriverManager
+ * @param timeoutMs a positive timeout in milliseconds
+ * @throws IllegalArgumentException if timeoutMs is not positive
+ */
+ public BoundedConnectionDriver(Driver delegate, long timeoutMs) {
+ this(delegate, timeoutMs, WORKERS);
+ }
+
+ BoundedConnectionDriver(Driver delegate, long timeoutMs, ExecutorService
workers) {
+ if (timeoutMs <= 0) {
+ throw new IllegalArgumentException("JDBC connect timeout must be
positive");
+ }
+ this.delegate = delegate;
+ this.timeoutMs = timeoutMs;
+ this.workers = workers;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @throws SQLTimeoutException if the connection deadline expires or the
caller is interrupted
+ * @throws SQLTransientConnectionException if all workers are busy or a
previous timed-out attempt
+ * through this driver is still running
+ */
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException {
+ if (abandonedAttempts.get() != 0) {
+ throw new SQLTransientConnectionException(
+ "A previous timed-out JDBC connection attempt is still running. Try
again after it finishes.",
+ "08001");
+ }
+ Attempt attempt = new Attempt();
+ ClassLoader context = Thread.currentThread().getContextClassLoader();
+ AccessControlContext callerContext = AccessController.getContext();
+ Properties properties = (Properties) info.clone();
+ try {
+ workers.execute(
+ () ->
+ AccessController.doPrivileged(
+ (PrivilegedAction<Void>)
+ () -> {
+ attempt.run(url, properties, context);
+ return null;
+ },
+ callerContext));
+ } catch (RejectedExecutionException e) {
+ throw new SQLTransientConnectionException(
+ "The JDBC connection attempt limit has been reached. Try again after
an attempt finishes.",
+ "08001",
+ e);
+ }
+ try {
+ Connection connection = attempt.result.get(timeoutMs,
TimeUnit.MILLISECONDS);
+ attempt.accept();
+ return connection;
+ } catch (TimeoutException e) {
+ attempt.abandon();
+ throw new SQLTimeoutException(
+ "JDBC connection attempt exceeded "
+ + timeoutMs
+ + " ms. Check backend and proxy connectivity.",
+ "08001",
+ e);
+ } catch (InterruptedException e) {
+ attempt.abandon();
+ Thread.currentThread().interrupt();
+ throw new SQLTimeoutException("Interrupted while opening a JDBC
connection", "08001", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof SQLException) {
+ throw (SQLException) cause;
+ }
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new SQLException("Failed to open JDBC connection", "08001", cause);
+ } finally {
+ attempt.decision.countDown();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean acceptsURL(String url) throws SQLException {
+ return delegate.acceptsURL(url);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info)
throws SQLException {
+ return delegate.getPropertyInfo(url, info);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int getMajorVersion() {
+ return delegate.getMajorVersion();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int getMinorVersion() {
+ return delegate.getMinorVersion();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean jdbcCompliant() {
+ return delegate.jdbcCompliant();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public java.util.logging.Logger getParentLogger() throws
SQLFeatureNotSupportedException {
+ return delegate.getParentLogger();
+ }
+
+ private final class Attempt {
+ private final CompletableFuture<Connection> result = new
CompletableFuture<>();
+ private final CountDownLatch decision = new CountDownLatch(1);
+ private boolean accepted;
+ private boolean abandoned;
+ private boolean finished;
+ @Nullable private Thread runner;
+
+ synchronized void accept() {
+ accepted = true;
+ }
+
+ synchronized void abandon() {
+ if (!finished) {
+ abandoned = true;
+ abandonedAttempts.incrementAndGet();
+ if (runner != null) {
+ runner.interrupt();
+ }
+ }
+ }
+
+ void run(String url, Properties properties, @Nullable ClassLoader context)
{
+ Thread current = Thread.currentThread();
+ ClassLoader previous = current.getContextClassLoader();
+ current.setContextClassLoader(context);
+ Connection connection = null;
+ try {
+ synchronized (this) {
+ runner = current;
+ if (abandoned) {
+ return;
+ }
+ }
+ connection = delegate.connect(url, properties);
+ result.complete(connection);
+ boolean interrupted = false;
+ while (true) {
+ try {
+ decision.await();
+ break;
+ } catch (InterruptedException e) {
+ // Cancellation must not bypass the handoff decision or discard a
late connection.
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ current.interrupt();
+ }
+ } catch (Throwable e) {
+ result.completeExceptionally(e);
+ } finally {
+ boolean delivered;
+ synchronized (this) {
+ delivered = accepted;
+ }
+ try {
+ if (!delivered && connection != null) {
+ try {
+ connection.close();
Review Comment:
Minor: after a timeout, the worker has its interrupt flag set when it gets
here, and a `close()` that honors the interrupt can fail. That failure is only
logged, the quarantine is cleared anyway (line 287), and the physical
connection can leak. Consider clearing the interrupt for the close and
restoring it afterwards.
Also, the `finally` only restores the context class loader (line 290). Any
`ThreadLocal`s the delegate driver sets, such as a Kerberos or Hadoop user,
stay on the shared worker, and the JVM-wide 64-thread pool (line 71) then runs
the next catalog's connect on that same thread. Could you document this, or
isolate workers per catalog?
##########
core/src/main/java/org/apache/gravitino/utils/BoundedConnectionDriver.java:
##########
@@ -0,0 +1,295 @@
+/*
+ * 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.utils;
+
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.security.ProtectionDomain;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Bounds physical connection creation for drivers that cannot enforce a login
timeout.
+ *
+ * <p>Only {@link Driver#connect} runs on a worker. Validation, metadata
queries and DDL stay on the
+ * caller. A connection that arrives after timeout or interruption is closed
by its worker and is
+ * never handed to a pool. Cancellation is best effort: a driver ignoring
interruption retains a
+ * worker until it returns. A JVM-wide limit of 64 workers, with no queue,
bounds that retention;
+ * excess attempts fail with {@link SQLTransientConnectionException}. Workers
are daemon threads and
+ * release their catalog context class loader after each attempt. Each
attempt, including late
+ * connection cleanup, runs with the submitting caller's access-control
context and JAAS subject;
+ * shared workers do not inherit the identity of the request that creates them.
+ *
+ * <p>After a timeout, further attempts through this instance fail immediately
until the abandoned
+ * attempt finishes cleanup. This prevents a lazy pool's initialization lock
from serializing a
+ * backlog of new attempts against the same unreachable backend. Normal pool
waits and SQL execution
+ * have their own budgets and are not bounded by this adapter.
+ */
+@SuppressWarnings("removal") // JDK 17 associates JAAS subjects with
access-control contexts.
+public final class BoundedConnectionDriver implements Driver {
+ private static final Logger LOG =
LoggerFactory.getLogger(BoundedConnectionDriver.class);
+
+ private static final AccessControlContext WORKER_CONTEXT =
+ new AccessControlContext(new ProtectionDomain[0]);
+
+ private static final ExecutorService WORKERS =
+ new ThreadPoolExecutor(
+ 0,
+ 64,
+ 60,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ task ->
+ AccessController.doPrivileged(
+ (PrivilegedAction<Thread>)
+ () -> {
+ Thread thread = new Thread(task,
"jdbc-connect-timeout");
+ thread.setDaemon(true);
+ thread.setContextClassLoader(
+ BoundedConnectionDriver.class.getClassLoader());
+ return thread;
+ },
+ WORKER_CONTEXT));
+
+ private final Driver delegate;
+ private final long timeoutMs;
+ private final ExecutorService workers;
+ private final AtomicInteger abandonedAttempts = new AtomicInteger();
+
+ /**
+ * Wraps a driver with a physical connection deadline.
+ *
+ * @param delegate the driver to call, without registering the wrapper with
DriverManager
+ * @param timeoutMs a positive timeout in milliseconds
+ * @throws IllegalArgumentException if timeoutMs is not positive
+ */
+ public BoundedConnectionDriver(Driver delegate, long timeoutMs) {
+ this(delegate, timeoutMs, WORKERS);
+ }
+
+ BoundedConnectionDriver(Driver delegate, long timeoutMs, ExecutorService
workers) {
+ if (timeoutMs <= 0) {
+ throw new IllegalArgumentException("JDBC connect timeout must be
positive");
+ }
+ this.delegate = delegate;
+ this.timeoutMs = timeoutMs;
+ this.workers = workers;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @throws SQLTimeoutException if the connection deadline expires or the
caller is interrupted
+ * @throws SQLTransientConnectionException if all workers are busy or a
previous timed-out attempt
+ * through this driver is still running
+ */
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException {
+ if (abandonedAttempts.get() != 0) {
+ throw new SQLTransientConnectionException(
+ "A previous timed-out JDBC connection attempt is still running. Try
again after it finishes.",
+ "08001");
+ }
+ Attempt attempt = new Attempt();
+ ClassLoader context = Thread.currentThread().getContextClassLoader();
+ AccessControlContext callerContext = AccessController.getContext();
+ Properties properties = (Properties) info.clone();
Review Comment:
Minor: `info.clone()` throws an NPE when `Driver.connect` is called with
null `Properties`, which JDBC allows. Please treat null as an empty
`Properties`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
