This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 11.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 912282a52da30016e286cbc767a4a9bf5c49d2a4 Author: opencode <[email protected]> AuthorDate: Wed Nov 4 16:25:13 2026 +0100 Close statement creations that land behind the return drain The tracker registers a statement after the driver has created it, while closeInvoked() closes the list the return finds and then stops reading it. A creation still in flight inside that window registered after the drain: it was not closed by that return, the next borrow's reset dropped the entry without closing the statement, and nothing ever closed it again - the statement stayed live on the very session being handed to the next borrower (verified with a real pool over H2; the pre-redesign code closed such statements at the next borrow). The list now carries a drained flag, set by closeInvoked() (including its empty-list early return) and cleared by reset() at the next borrow and guarded by statementsLock. A registration that arrives drained does not enter the list: createStatement closes the statement where it arrives, so no open statement can survive the return on a reused session. Two regression tests pin the interleaving, one at the hook level and one through the full chain and a real driver by parking the return inside the tracker's own close of a leaked statement while a second thread's prepare completes; both fail without the fix. --- .../jdbc/pool/interceptor/StatementFinalizer.java | 44 +++- .../pool/interceptor/TestStatementFinalizer.java | 281 +++++++++++++++++++++ 2 files changed, 320 insertions(+), 5 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementFinalizer.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementFinalizer.java index 303d132291..ee8a45c711 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementFinalizer.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementFinalizer.java @@ -48,7 +48,9 @@ import org.apache.tomcat.jdbc.pool.util.MessageFactory; * close meets a statement already closed as a harmless no-op, as they were before. An interceptor placed outside * this one could swallow a close in the same way, but the pool's own {@link StatementCache} no longer does: listed * above this interceptor it refuses to recycle, because a recycled statement is left open rather than closed and - * would be reclaimed here alive, out from under the cache.) + * would be reclaimed here alive, out from under the cache.) A statement whose driver create was still in flight + * while the connection was returned cannot be reclaimed at the return either - it registers after that return has + * drained the list - so such a late registration is closed where it arrives instead of being tracked. * <p> * Tracked statements are handed to the application as proxies implementing the standard JDBC statement interfaces, as * is the statement facade the pool always installs in front of this interceptor and as are the proxies of the other @@ -68,12 +70,24 @@ public class StatementFinalizer extends AbstractCreateStatementInterceptor { protected final List<StatementEntry> statements = new LinkedList<>(); /** - * Monitor guarding {@link #statements}. The pool's cleaner thread may clear the list (through {@link #reset}) while - * a borrower is still creating or closing statements on an abandoned connection, so subclasses that access the list - * must hold this lock as well. + * Monitor guarding {@link #statements} and {@link #returnDrained}. The pool's cleaner thread may clear the list + * (through {@link #reset}) while a borrower is still creating or closing statements on an abandoned connection, so + * subclasses that access either must hold this lock as well. */ protected final Object statementsLock = new Object(); + /** + * Whether the connection return has already drained {@link #statements}: set by {@link #closeInvoked()}, cleared by + * {@link #reset} at the start of the next borrow. Every access is guarded by {@link #statementsLock}. + * <p> + * A statement creation is not serialized against the connection return: the driver's create has completed and the + * statement is open before this interceptor can register it. A creation whose connection is returned inside that + * window would register after the drain - not closed by that return, dropped by the next borrow's {@link #reset}, + * and left open on a session that is about to serve another borrower. A registration that arrives drained + * therefore does not enter the list: {@link #createStatement} closes the statement itself. + */ + protected boolean returnDrained = false; + /** * Default constructor. */ @@ -105,8 +119,20 @@ public class StatementFinalizer extends AbstractCreateStatementInterceptor { return tracked; } StatementEntry entry = new StatementEntry(tracked); + boolean drained; synchronized (statementsLock) { - statements.add(entry); + drained = returnDrained; + if (!drained) { + statements.add(entry); + } + } + if (drained) { + // The connection return drained the list before this statement + // could be registered, so the return's renewed close cannot find + // it; close it here rather than leave it open on a session that + // is about to serve the next borrower. + entry.close(); + return tracked; } try { return getConstructor(constructorIndex, statementInterface) @@ -124,6 +150,10 @@ public class StatementFinalizer extends AbstractCreateStatementInterceptor { public void closeInvoked() { List<StatementEntry> unclosed; synchronized (statementsLock) { + // From this point a creation still in flight can no longer be + // reclaimed by this drain; it is told to close itself instead + // (see returnDrained). + returnDrained = true; if (statements.isEmpty()) { return; } @@ -150,6 +180,10 @@ public class StatementFinalizer extends AbstractCreateStatementInterceptor { @Override public void reset(ConnectionPool parent, PooledConnection con) { synchronized (statementsLock) { + // A new borrow starts here (or the instance goes out of + // service): statements created from now on are accepted into + // the list again. + returnDrained = false; statements.clear(); } super.reset(parent, con); diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementFinalizer.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementFinalizer.java index 2be6079dc4..ec02884ecc 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementFinalizer.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementFinalizer.java @@ -17,6 +17,10 @@ package org.apache.tomcat.jdbc.pool.interceptor; import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; import java.sql.CallableStatement; import java.sql.Connection; import java.sql.PreparedStatement; @@ -26,12 +30,19 @@ import java.sql.Statement; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.apache.tomcat.jdbc.JdbcPoolBaseTest; +import org.apache.tomcat.jdbc.pool.ConnectionPool; import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.JdbcInterceptor; +import org.apache.tomcat.jdbc.pool.PooledConnection; /** * What {@code StatementFinalizer} owes the application from the other @@ -383,4 +394,274 @@ class TestStatementFinalizer extends JdbcPoolBaseTest { Assertions.assertEquals(0, tracked(inner), "the inner tracker must forget its entry"); } } + + /* + * A statement handle the test holds while pinning whether the tracker + * ever asked its driver object to close. + */ + private static Statement fakedStatement() { + final AtomicBoolean closed = new AtomicBoolean(); + return (Statement) Proxy.newProxyInstance( + TestStatementFinalizer.class.getClassLoader(), + new Class<?>[] { Statement.class }, + (p, m, a) -> { + switch (m.getName()) { + case "close": + closed.set(true); + return null; + case "isClosed": + return Boolean.valueOf(closed.get()); + case "equals": + return Boolean.valueOf(p == (a == null || a.length == 0 ? null : a[0])); + case "hashCode": + return Integer.valueOf(System.identityHashCode(p)); + default: + return null; + } + }); + } + + private static boolean closeWasAsked(Statement fake) { + // The fake records the close on the object itself; re-reading + // isClosed() through the handle is what the interceptor would see. + try { + return fake.isClosed(); + } catch (SQLException x) { + throw new IllegalStateException(x); + } + } + + @Test + void aStatementRegisteringAfterTheReturnHasDrainedIsClosedWhereItArrives() throws Exception { + // The interleaving the race produces, driven at the hook level so + // it is deterministic: the driver create has completed (the + // statement is open in the hand of the tracker), the connection + // return drains the list before the registration lands, and the + // late entry must not sit on a list that no return will ever read + // again. + StatementFinalizer finalizer = new StatementFinalizer(); + Method create = Connection.class.getMethod("createStatement"); + + finalizer.reset(null, null); // the borrow starts, creating is accepted + finalizer.closeInvoked(); // the return drains the (empty) list + + Statement late = fakedStatement(); + Object handle = finalizer.createStatement(null, create, null, late, 0L); + + Assertions.assertTrue(closeWasAsked(late), + "a create that raced the return must be closed where it registers - it would otherwise " + + "survive untracked on a session the pool hands to the next borrower"); + Assertions.assertSame(late, handle, + "nothing was tracked, so the handle is the statement itself"); + Assertions.assertEquals(0, tracked(finalizer), + "the closed late statement must not enter the list"); + + // The next borrow re-arms the tracker: ordinary creates are tracked + // and reclaimed at the return again. + Statement fresh = fakedStatement(); + finalizer.reset(null, null); + finalizer.createStatement(null, create, null, fresh, 0L); + Assertions.assertEquals(1, tracked(finalizer), "creating must be accepted again at the next borrow"); + finalizer.closeInvoked(); + Assertions.assertTrue(closeWasAsked(fresh), "the return must still reclaim ordinary statements"); + Assertions.assertEquals(0, tracked(finalizer)); + } + + @Test + void aCreateStillInFlightWhileTheConnectionIsReturnedDoesNotReachTheNextBorrower() throws Exception { + // The same interleaving through the whole chain and a real driver: + // a leaked statement parks the return inside the close that the + // tracker itself issues, and while the return is held there - after + // its drain, before the session is handed back - another thread's + // in-flight create lands. + DataSource ds = finalizerPool(); + ds.getPoolProperties().setJdbcInterceptors(StatementFinalizer.class.getName() + ';' + + StatementGateInterceptor.class.getName()); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setMaxIdle(1); + ds.createPool(); + + Connection first = ds.getConnection(); + setSessionMarker(first, 11); + StatementFinalizer finalizer = interceptorOf(first, StatementFinalizer.class); + Assertions.assertNotNull(finalizer); + + StatementGateInterceptor.holdFirstClose(); + Statement leaked = first.createStatement(); // its return-time close is the parking post + + CountDownLatch createEntered = new CountDownLatch(1); + CountDownLatch createRelease = new CountDownLatch(1); + StatementGateInterceptor.holdCreateOf("in-flight-prepare", createEntered, createRelease); + + final AtomicReference<PreparedStatement> holder = new AtomicReference<>(); + final AtomicReference<Throwable> failure = new AtomicReference<>(); + Thread creator = new Thread(() -> { + try { + holder.set(first.prepareStatement("SELECT 42")); + } catch (Throwable x) { + failure.set(x); + } + }, "in-flight-prepare"); + + final AtomicReference<Throwable> closeFailure = new AtomicReference<>(); + CountDownLatch closeParked = new CountDownLatch(1); + CountDownLatch closeRelease = new CountDownLatch(1); + StatementGateInterceptor.parkFirstCloseOnReturn(closeParked, closeRelease); + Thread closer = new Thread(() -> { + try { + first.close(); + } catch (Throwable x) { + closeFailure.set(x); + } + }, "returning-thread"); + + creator.start(); + Assertions.assertTrue(createEntered.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + "the prepare is in flight inside the chain"); + closer.start(); + Assertions.assertTrue(closeParked.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + "the return is parked inside the tracker's close of the leaked statement"); + + createRelease.countDown(); + creator.join(AWAIT_TIMEOUT_MS); + Assertions.assertFalse(creator.isAlive(), "the prepare completed within the timeout"); + Assertions.assertNull(failure.get(), "the in-flight create must not fail: " + failure.get()); + Assertions.assertEquals(0, tracked(finalizer), + "the create that landed behind the drain must not sit on the tracker - no later return " + + "will ever read it"); + + closeRelease.countDown(); + closer.join(AWAIT_TIMEOUT_MS); + Assertions.assertFalse(closer.isAlive(), "the return completed within the timeout"); + Assertions.assertNull(closeFailure.get(), "the return must not fail: " + closeFailure.get()); + StatementGateInterceptor.disarm(); + + PreparedStatement inFlight = holder.get(); + Assertions.assertNotNull(inFlight); + Assertions.assertTrue(inFlight.isClosed(), + "the statement that raced the return must be closed where it registers, not left " + + "open on the session that goes back to the pool"); + Assertions.assertEquals(0, tracked(finalizer), + "nothing may be left on the tracker after the return"); + + Connection second = ds.getConnection(); + try { + Assertions.assertEquals(Integer.valueOf(11), sessionMarker(second), + "precondition: the same physical session serves the next borrower"); + Assertions.assertEquals(0, tracked(interceptorOf(second, StatementFinalizer.class)), + "the next borrow must start with an empty tracker"); + try (ResultSet rs = inFlight.executeQuery()) { + Assertions.fail("the leaked statement still executes: " + (rs.next() ? rs.getInt(1) : "?")); + } catch (SQLException expected) { + // closed statements refuse use - the point of the assertion + } + } finally { + second.close(); + } + } + + /** + * Statement gate for the race test: wraps every statement the chain + * creates so the test can park a close issued from inside the pool + * (the tracker's own return-time close), and can hold one named + * thread's statement creation in flight while the return is parked. + * Armed through statics because the pool instantiates configured + * interceptors itself; hence public static with a no-arg constructor. + */ + public static class StatementGateInterceptor extends JdbcInterceptor { + + private static final AtomicBoolean holdFirstCloseArmed = new AtomicBoolean(); + private static volatile CountDownLatch closeParked; + private static volatile CountDownLatch closeRelease; + private static final AtomicBoolean closeHoldTaken = new AtomicBoolean(); + private static volatile String createHoldThread; + private static volatile CountDownLatch createEntered; + private static volatile CountDownLatch createRelease; + + static void holdFirstClose() { + closeHoldTaken.set(false); + holdFirstCloseArmed.set(true); + } + + static void parkFirstCloseOnReturn(CountDownLatch parked, CountDownLatch release) { + closeParked = parked; + closeRelease = release; + } + + static void holdCreateOf(String threadName, CountDownLatch entered, CountDownLatch release) { + createHoldThread = threadName; + createEntered = entered; + createRelease = release; + } + + static void disarm() { + closeParked = null; + closeRelease = null; + createHoldThread = null; + createEntered = null; + createRelease = null; + holdFirstCloseArmed.set(false); + closeHoldTaken.set(false); + } + + @Override + public void reset(ConnectionPool parent, PooledConnection con) { + // no per-borrow state + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + String name = method.getName(); + boolean isCreate = "createStatement".equals(name) || "prepareStatement".equals(name) + || "prepareCall".equals(name); + CountDownLatch entered = createEntered; + if (isCreate && entered != null && Thread.currentThread().getName().equals(createHoldThread)) { + entered.countDown(); + CountDownLatch release = createRelease; + if (release != null) { + release.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + } + Object result = super.invoke(proxy, method, args); + if (isCreate && result instanceof Statement) { + boolean parkItsClose = holdFirstCloseArmed.get() && closeHoldTaken.compareAndSet(false, true); + Statement created = (Statement) result; + Class<?> iface = created instanceof CallableStatement ? CallableStatement.class + : created instanceof PreparedStatement ? PreparedStatement.class : Statement.class; + return Proxy.newProxyInstance(StatementGateInterceptor.class.getClassLoader(), + new Class<?>[] { iface }, new StatementGate(created, parkItsClose)); + } + return result; + } + + static class StatementGate implements InvocationHandler { + private final Statement delegate; + private final boolean parkClose; + private final AtomicBoolean closeParkedOnce = new AtomicBoolean(); + + StatementGate(Statement delegate, boolean parkClose) { + this.delegate = delegate; + this.parkClose = parkClose; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if ("close".equals(method.getName()) && parkClose && closeParkedOnce.compareAndSet(false, true)) { + CountDownLatch parked = closeParked; + CountDownLatch release = closeRelease; + if (parked != null) { + parked.countDown(); + if (release != null) { + release.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + } + } + try { + return method.invoke(delegate, args); + } catch (InvocationTargetException x) { + throw x.getCause() != null ? x.getCause() : x; + } + } + } + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
