yuqi1129 commented on code in PR #14508:
URL: https://github.com/apache/gravitino/pull/14508#discussion_r4465936275
##########
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:
Addressed in `63e5a60dd9`: cold table loads preserve NoSuchSchemaException
instead of converting it to NoSuchTableException. The API contract and
Java/Python existence helpers were updated to treat either missing schema or
missing table as false. The dispatcher and REST table suites passed again in
this round.
##########
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:
Addressed in `63e5a60dd9`: null connection properties become an empty
Properties instance; non-null input is cloned. All 15 bounded-driver tests,
including the null-properties case, passed again in this round.
##########
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:
The current code clears the interrupt flag before late close, then restores
it after cleanup (including an interrupt restored by the delegate); the
interrupt-sensitive-close regression passed again. The class Javadoc and
docs/jdbc-catalog-connection-validation.md also explicitly limit isolation to
access-control/JAAS contexts: arbitrary ThreadLocal/InheritableThreadLocal
state is not cleaned, delegates must clean it and drivers relying on it must
not use this opt-in shared adapter. Per-catalog workers alone would not isolate
different users of the same catalog, so no broader isolation guarantee is
claimed.
##########
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:
For clarity after the follow-up review: the current code deliberately
restores the interrupt flag only after cleanup, as requested in the latest
review. ThreadPoolExecutor still clears it before the next task; this does not
propagate cancellation into the next connection attempt. Clearing the flag
around close is the essential resource-cleanup fix, and the late-close
regression verifies that behavior.
--
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]
