This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit d29ea415fdd4ca7e1a54763b8d8de0dec0f07368 Author: opencode <[email protected]> AuthorDate: Tue Oct 27 23:38:36 2026 +0100 Deliver pool JMX notifications off the pool threads The pool built each notification and dispatched it synchronously on the thread that reported the condition. Several of the notifications are raised from the connection lifecycle with the lock of the connection concerned held - the abandoned, suspect and suspect-returned reports - and one, POOL EMPTY, is raised once per failed borrow from an exhausted pool. A monitoring listener that was slow, or that blocked on anything from a saturated log to a stalled socket, therefore held up the borrow, return or abandon processing of that connection for exactly as long as it held on, and repeated the cost on every miss while the pool stayed exhausted. Delivery now runs on a dedicated notifier thread: the MBean hands the JMX listeners to a single-worker executor through the NotificationBroadcasterSupport(Executor) constructor, and notify() submits the locally registered listeners to the same executor behind the JMX handoff. The notification is still built on the calling thread, so it carries the time, sequence number and source of the condition it reports, and the single worker keeps the notifications in the order the pool raised them. The delivery queue is bounded; once a listener stops draining it, further notifications are dropped and logged rather than delaying the pool, and the notifier thread is a daemon that starts on the first notification and dies when idle, so a pool that never notifies costs nothing. The result of notify() now reports the handoff only. It no longer says anything about whether a listener was reached or what it did with the notification; in particular it no longer returns false because a local listener threw, since by the time a listener runs the report has already been accepted. Tested by new TestJmxNotificationDelivery: a listener stuck on a latch no longer delays the thread that reported - on a failed borrow from an exhausted pool either - a burst arrives in the order it was reported, and a delivery queue nobody drains drops without failing the pool. TestJmxNotificationSource now waits for delivery with the base test's bounded await before making the same assertions as before. Full jdbc-pool suite passes 302 tests. --- .../tomcat/jdbc/pool/jmx/ConnectionPool.java | 105 +++++++- .../jdbc/pool/jmx/TestJmxNotificationDelivery.java | 281 +++++++++++++++++++++ .../jdbc/pool/jmx/TestJmxNotificationSource.java | 9 + webapps/docs/changelog.xml | 17 ++ 4 files changed, 403 insertions(+), 9 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java index a9597793ff..19157768a5 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/jmx/ConnectionPool.java @@ -17,6 +17,10 @@ package org.apache.tomcat.jdbc.pool.jmx; import java.util.Properties; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import javax.management.MBeanNotificationInfo; @@ -62,6 +66,42 @@ public class ConnectionPool extends NotificationBroadcasterSupport protected ConcurrentLinkedQueue<NotificationListener> listeners = new ConcurrentLinkedQueue<>(); + /** + * How many delivery tasks the notifier thread accepts into its queue + * before further notifications are dropped and logged rather than + * queued. The queue bounds the memory a flooded event source - the pool + * notifying once per failed borrow while it is exhausted - can make the + * pool allocate; it is not a promise that every notification is + * delivered. + */ + private static final int NOTIFY_QUEUE_CAPACITY = 2000; + + /** + * The single thread that pool notifications are delivered on. Delivery + * goes through here rather than running on the thread that reports the + * condition, because several of the pool's notifications are raised from + * the connection lifecycle - some of them while holding the lock of the + * connection concerned - and a monitoring listener that is slow or + * blocked must not stall those paths. One worker keeps the notifications + * in the order the pool raised them; the bounded queue and the drop on + * overflow keep a listener that stops draining from growing the queue + * without limit. The thread is a daemon, is started on the first + * notification and dies when it has been idle, so a pool that never + * notifies costs nothing, and it never holds up the JVM on shutdown. + * Shared by every pool MBean in the JVM: notifications of one pool can + * queue behind those of another, but a blocked listener then only + * starves delivery - the drop above is what a pool's operations feel, + * not the block. + */ + private static final ThreadPoolExecutor NOTIFIER = new ThreadPoolExecutor( + 0, 1, 60L, TimeUnit.SECONDS, + new LinkedBlockingQueue<Runnable>(NOTIFY_QUEUE_CAPACITY), + r -> { + Thread t = new Thread(r, "JDBC-Pool-JMX-Notifier"); + t.setDaemon(true); + return t; + }); + /** * the ObjectName of this pool. */ @@ -73,7 +113,9 @@ public class ConnectionPool extends NotificationBroadcasterSupport * @param pool the connection pool to wrap */ public ConnectionPool(org.apache.tomcat.jdbc.pool.ConnectionPool pool) { - super(); + // hand the JMX listener delivery to the notifier thread as well; + // see notify(String,String) + super(NOTIFIER); this.pool = pool; } @@ -189,15 +231,29 @@ public class ConnectionPool extends NotificationBroadcasterSupport } /** - * Return true if the notification was sent successfully, false otherwise. + * Reports a condition to the pool's notification listeners. * <p> - * The notification is sent with the pool's {@link ObjectName} as its source - * - see {@link #getNotificationSource()} - so that a client which receives - * the notification over Java serialization can receive it at all. + * The delivery is asynchronous: both the listeners reached through JMX - + * delivered by {@link #sendNotification(Notification)} - and the ones + * registered locally with {@link #addListener(NotificationListener)} + * receive the notification on the pool's notification thread, not on the + * thread that calls this method. The pool raises several of its + * notifications from the connection lifecycle, some of them while holding + * the lock of the connection concerned, and a monitoring listener that is + * slow or blocked must not reach those paths. The notification itself is + * still built on the calling thread, so it carries the time, sequence + * number and serializable source of the condition it reports - see + * {@link #getNotificationSource()} - and, since the notifier is a single + * thread, listeners observe notifications in the order the pool reported + * them. * * @param type The notification type * @param message The message - * @return true if the notification succeeded + * @return true if the notification was accepted for delivery, false if it + * was dropped - which happens when the delivery queue is full or + * the notification could not be built. The result reports the + * handoff only: it says nothing about whether any listener was + * reached or what a listener did with the notification. */ public boolean notify(final String type, String message) { try { @@ -207,11 +263,21 @@ public class ConnectionPool extends NotificationBroadcasterSupport sequence.incrementAndGet(), System.currentTimeMillis(), "["+type+"] "+message); + // queues the notification for the JMX listeners on the notifier sendNotification(n); - for (NotificationListener listener : listeners) { - listener.handleNotification(n,this); - } + // and the same for the local listeners, behind the JMX handoff so + // that per notification the listener order stays as it was when + // delivery was synchronous + NOTIFIER.execute(() -> deliverLocally(n, type)); return true; + }catch (RejectedExecutionException x) { + // the delivery queue is full; dropping the notification is the + // point of the bound, and must not fail the pool operation that + // reported the condition + log.warn("Dropping notification of type ["+type+"] for pool [" + +getNotificationSource()+"]: the notification delivery queue (" + +NOTIFY_QUEUE_CAPACITY+" deliveries) is full"); + return false; }catch (Exception x) { if (log.isDebugEnabled()) { log.debug("Notify failed. Type="+type+"; Message="+message,x); @@ -221,6 +287,27 @@ public class ConnectionPool extends NotificationBroadcasterSupport } + /** + * Hands a notification to the listeners registered with this MBean + * outside JMX. Runs on the notifier thread. A listener that throws is + * contained here: its exception may not kill the notifier thread that + * every pool in the JVM delivers its notifications on. + * + * @param n the notification to deliver + * @param type the notification type, for the failure log + */ + private void deliverLocally(Notification n, String type) { + try { + for (NotificationListener listener : listeners) { + listener.handleNotification(n, this); + } + } catch (Exception x) { + if (log.isDebugEnabled()) { + log.debug("Local notification listener failed. Type="+type,x); + } + } + } + /** * Returns the source the pool's notifications are sent with. * <p> diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java new file mode 100644 index 0000000000..5d3d88a31a --- /dev/null +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationDelivery.java @@ -0,0 +1,281 @@ +/* + * 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.tomcat.jdbc.pool.jmx; + +import java.sql.Connection; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import javax.management.Notification; +import javax.management.NotificationListener; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.tomcat.jdbc.JdbcPoolBaseTest; +import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.PoolExhaustedException; + +/** + * Pool notifications are delivered off the thread that reports the condition. + * <p> + * Several of the pool's notifications are raised from the connection + * lifecycle, some of them while the pool holds the lock of the connection + * concerned - the abandon, suspect and suspect-returned paths and the + * connect-failure path - and a failed borrow on an exhausted pool reports + * POOL_EMPTY once per miss. When delivery ran on the calling thread, one slow + * or blocked monitoring listener therefore stalled connection lifecycle + * operations and amplified the load precisely when the pool was saturated. + * Delivery now goes through the pool's single notifier thread, and a + * notification that no longer fits the bounded delivery queue is dropped and + * logged rather than queued or thrown into the reporting thread.</p> + * <p> + * The stuck listener used below always bounds its own wait, so a regression + * to synchronous delivery fails these tests on the timeout bound instead of + * hanging the suite.</p> + */ +class TestJmxNotificationDelivery extends JdbcPoolBaseTest { + + /* + * Upper bound on how long the thread that reports a condition may take, + * measured against a listener that stays in handleNotification for + * STUCK_LISTENER_TIMEOUT_S. Far below the stuck interval on a healthy + * handoff, at or above it on a delivery that runs on the caller. + */ + private static final long CALLER_MUST_RETURN_WITHIN_MS = 5000; + + /* How long the stuck listener holds on at most; bounds the stuck state. */ + private static final long STUCK_LISTENER_TIMEOUT_S = 10; + + /* How many notifications are pushed at a queue nobody is draining. */ + private static final int FLOOD_SIZE = 4000; + + /** + * Every notification type the pool MBean advertises, in the order the + * burst test reports them. + */ + private static final String[] ALL_TYPES = { + ConnectionPool.NOTIFY_INIT, + ConnectionPool.NOTIFY_CONNECT, + ConnectionPool.NOTIFY_ABANDON, + ConnectionPool.SLOW_QUERY_NOTIFICATION, + ConnectionPool.FAILED_QUERY_NOTIFICATION, + ConnectionPool.SUSPECT_ABANDONED_NOTIFICATION, + ConnectionPool.POOL_EMPTY, + ConnectionPool.SUSPECT_RETURNED_NOTIFICATION }; + + /* + * A listener that records the notifications it receives, signals that it + * has been reached, stays inside handleNotification until the test + * releases it - and never longer than STUCK_LISTENER_TIMEOUT_S, so that a + * delivery running on the caller thread of notify() stalls for that bound + * and no further. + */ + private static final class StuckListener implements NotificationListener { + private final CountDownLatch entered = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final CountDownLatch completed = new CountDownLatch(1); + private final List<Notification> received = new CopyOnWriteArrayList<>(); + + @Override + public void handleNotification(Notification notification, Object handback) { + received.add(notification); + entered.countDown(); + try { + release.await(STUCK_LISTENER_TIMEOUT_S, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + completed.countDown(); + } + } + + @Test + void aStuckLocalListenerDoesNotStallTheThreadThatReports() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setName("JmxDeliveryStuck-" + System.nanoTime()); + ds.createPool(); + // an unregistered MBean is enough: this is about the local fan-out + ConnectionPool jmxPool = new ConnectionPool(ds.getPool()); + StuckListener stuck = new StuckListener(); + jmxPool.addListener(stuck); + try { + long start = System.currentTimeMillis(); + boolean reported = jmxPool.notify(ConnectionPool.NOTIFY_ABANDON, + "a connection was abandoned"); + long elapsedMs = System.currentTimeMillis() - start; + + Assertions.assertTrue(reported, + "the notification must be reported as handed off for delivery"); + Assertions.assertTrue(stuck.entered.await(5, TimeUnit.SECONDS), + "the notification must still reach the listener"); + Assertions.assertTrue(elapsedMs < CALLER_MUST_RETURN_WITHIN_MS, + "reporting a condition must not wait for the listener: notify() took " + + elapsedMs + " ms, of the " + STUCK_LISTENER_TIMEOUT_S + + " s the listener holds on, so the delivery ran on the" + + " thread that reported the condition"); + Assertions.assertEquals(1, stuck.completed.getCount(), + "the listener must still be stuck at the point the caller was released"); + } finally { + stuck.release.countDown(); + } + Assertions.assertTrue(stuck.completed.await(5, TimeUnit.SECONDS), + "the released listener must have finished handling the notification"); + jmxPool.removeListener(stuck); + } + + @Test + void aFailedBorrowOnAnExhaustedPoolReportsWithoutStallingTheBorrowThread() + throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setJmxEnabled(true); + ds.getPoolProperties().setName("JmxDeliveryEmpty-" + System.nanoTime()); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setInitialSize(1); + // a small positive wait, so an exhausted pool reports the empty pool + // once its borrower gives up - note that maxWait<=0 means to wait + // forever, not to wait not at all, so 0 here would never time out + ds.getPoolProperties().setMaxWait(250); + ds.createPool(); + ConnectionPool jmxPool = ds.getPool().getJmxPool(); + Assertions.assertNotNull(jmxPool, "the fixture must have created the pool MBean"); + StuckListener stuck = new StuckListener(); + jmxPool.addListener(stuck); + long elapsedMs; + try (Connection held = ds.getConnection()) { + PoolExhaustedException exhausted = null; + long start = System.currentTimeMillis(); + try { + ds.getConnection(); + } catch (PoolExhaustedException expected) { + exhausted = expected; + } + elapsedMs = System.currentTimeMillis() - start; + + Assertions.assertNotNull(exhausted, + "a borrow that times out on an exhausted pool must fail"); + Assertions.assertTrue(stuck.entered.await(5, TimeUnit.SECONDS), + "the failed borrow must still report POOL EMPTY"); + Assertions.assertTrue(elapsedMs < CALLER_MUST_RETURN_WITHIN_MS, + "the failed borrow must not wait for the listener: it took " + elapsedMs + + " ms, of the " + STUCK_LISTENER_TIMEOUT_S + " s the listener" + + " holds on, so the notification ran on the borrow thread"); + Assertions.assertEquals(1, stuck.completed.getCount(), + "the listener must still be stuck at the point the borrow returned"); + } finally { + stuck.release.countDown(); + } + awaitUntil(() -> !stuck.received.isEmpty()); + Assertions.assertEquals(ConnectionPool.POOL_EMPTY, stuck.received.get(0).getType(), + "what the failed borrow reported must be the pool-empty notification"); + jmxPool.removeListener(stuck); + } + + /* + * The order the pool reported its notifications in is the order the + * listeners must observe them in, across a burst covering every type. A + * contract pin rather than a regression pin: a synchronous fan-out also + * delivers in order, and the fix must not lose that. + */ + @Test + void everyTypeOfABurstArrivesInReportingOrder() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setName("JmxDeliveryBurst-" + System.nanoTime()); + ds.createPool(); + ConnectionPool jmxPool = new ConnectionPool(ds.getPool()); + List<Notification> received = new CopyOnWriteArrayList<>(); + NotificationListener listener = (notification, handback) -> received.add(notification); + jmxPool.addListener(listener); + try { + for (String type : ALL_TYPES) { + Assertions.assertTrue(jmxPool.notify(type, "burst " + type), + "the notification of type [" + type + "] must be reported as sent"); + } + + awaitUntil(() -> received.size() >= ALL_TYPES.length); + Assertions.assertEquals(ALL_TYPES.length, received.size(), + "every type of the burst must have reached the listener"); + long previousSequence = -1; + for (int i = 0; i < ALL_TYPES.length; i++) { + Notification notification = received.get(i); + Assertions.assertEquals(ALL_TYPES[i], notification.getType(), + "position " + i + " of the burst must be the type reported at" + + " position " + i + ", delivery must not reorder"); + Assertions.assertTrue(notification.getSequenceNumber() > previousSequence, + "sequence numbers must keep increasing across the burst, got " + + notification.getSequenceNumber() + " after " + + previousSequence); + previousSequence = notification.getSequenceNumber(); + } + } finally { + jmxPool.removeListener(listener); + } + } + + @Test + void aDeliveryQueueNobodyDrainsDropsInsteadOfStallingOrFailingThePool() + throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setName("JmxDeliveryOverflow-" + System.nanoTime()); + ds.createPool(); + ConnectionPool jmxPool = new ConnectionPool(ds.getPool()); + StuckListener stuck = new StuckListener(); + jmxPool.addListener(stuck); + int handedOff = 0; + int dropped = 0; + try { + for (int i = 0; i < FLOOD_SIZE; i++) { + long start = System.currentTimeMillis(); + boolean reported; + try { + reported = jmxPool.notify(ConnectionPool.POOL_EMPTY, "overflow " + i); + } catch (RuntimeException x) { + throw new AssertionError("notify() must not throw into the thread that" + + " reports the condition", x); + } + long elapsedMs = System.currentTimeMillis() - start; + Assertions.assertTrue(elapsedMs < CALLER_MUST_RETURN_WITHIN_MS, + "reporting a condition onto an undrainable queue must not block:" + + " notify() call " + i + " took " + elapsedMs + " ms"); + if (reported) { + handedOff++; + } else { + dropped++; + } + } + + Assertions.assertTrue(stuck.entered.await(5, TimeUnit.SECONDS), + "the first notification must have reached the listener"); + // the queue is bounded, so a listener that does not drain it + // cannot make the pool accept the entire flood + Assertions.assertTrue(dropped > 0, + "a queue that nothing drains must start dropping once full: all " + + FLOOD_SIZE + " notifications were accepted, so the queue" + + " is unbounded"); + Assertions.assertTrue(handedOff > 0, + "before the queue is full, the notifications must be handed off"); + } finally { + stuck.release.countDown(); + } + awaitUntil(() -> !stuck.received.isEmpty()); + Assertions.assertEquals(ConnectionPool.POOL_EMPTY, stuck.received.get(0).getType(), + "the first notification must have been delivered, not dropped"); + jmxPool.removeListener(stuck); + } +} diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationSource.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationSource.java index 46a3bec781..3a0f712f10 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationSource.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/jmx/TestJmxNotificationSource.java @@ -57,6 +57,12 @@ import org.apache.tomcat.jdbc.pool.DataSource; * rewriting is also what a client behind a connector effectively sees, which is * why the last test pins that the name the remote side reads does not change * with this fix.</p> + * <p> + * Delivery is asynchronous (on the pool's notification thread), so the tests + * wait for the expected notifications to arrive with the bounded + * {@link JdbcPoolBaseTest#awaitUntil(java.util.function.BooleanSupplier)} + * before asserting; once they have all arrived, every pin below is asserted + * exactly as if delivery had been immediate.</p> */ class TestJmxNotificationSource extends JdbcPoolBaseTest { @@ -127,6 +133,7 @@ class TestJmxNotificationSource extends JdbcPoolBaseTest { "the notification of type [" + type + "] must be reported as sent"); } + awaitUntil(() -> received.size() >= ALL_TYPES.length); Assertions.assertEquals(ALL_TYPES.length, received.size(), "every type must have reached the listener"); long previousSequence = -1; @@ -185,6 +192,7 @@ class TestJmxNotificationSource extends JdbcPoolBaseTest { "the notification of type [" + type + "] must be reported as sent"); } + awaitUntil(() -> received.size() >= ALL_TYPES.length); Assertions.assertEquals(ALL_TYPES.length, received.size(), "every type must have reached the listener"); for (int i = 0; i < ALL_TYPES.length; i++) { @@ -232,6 +240,7 @@ class TestJmxNotificationSource extends JdbcPoolBaseTest { Assertions.assertTrue(jmxPool.notify(ConnectionPool.POOL_EMPTY, "pool is empty"), "the notification must be reported as sent"); + awaitUntil(() -> !received.isEmpty()); Assertions.assertEquals(1, received.size(), "a client subscribed through the MBean server must be reached"); Notification notification = received.get(0); diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index f2e4738f2d..9d61b54f94 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -925,6 +925,23 @@ connection is now <code>-1</code> instead of no value at all. (remm) </fix> + <fix> + Pool JMX notifications are now delivered on a dedicated + notification thread instead of on the thread that reported the + condition. Several of the pool's notifications are raised from + the connection lifecycle, some of them while the pool holds the + lock of the connection concerned, so a monitoring listener that + was slow or blocked stalled the borrow, return and abandon + processing of that connection for as long as it held on, and a + failed borrow from an exhausted pool repeated that cost on every + miss. Listeners observe the notifications in the same order the + pool reported them; a listener that stops draining the bounded + delivery queue now causes later notifications to be dropped and + logged rather than delaying the pool. The result of reporting a + notification through the pool MBean now describes that handoff + only, and no longer depends on whether a locally registered + listener threw. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
