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 47577409560c319f58fe49cda4c7b96a858fa8d7 Author: opencode <[email protected]> AuthorDate: Thu Oct 29 23:27:18 2026 +0100 FairBlockingQueue: restore the interrupt status when a poll is interrupted but an item is delivered When a timed poll's wait was interrupted but an offer had committed (or concurrently committed) the item to the waiter's latch, the queue delivered the item and called Thread.interrupted(), clearing the interrupt flag and propagating nothing. An interrupt meant for a later phase - container stop, executor shutdown - was silently lost, and the pool's propagateInterruptState policy was bypassed because the normal return path throws nothing. The delivered item is kept and the interrupt status is now re-asserted before returning, the same contract the JDK's hand-off queues follow. The queue's fast path already returned items with the flag set, so borrowers must already tolerate a pending interrupt after a successful borrow. Adds a deterministic test pinning delivery and flag restoration. --- .../apache/tomcat/jdbc/pool/FairBlockingQueue.java | 14 +++-- .../tomcat/jdbc/pool/TestFairBlockingQueue.java | 69 +++++++++++++++++++++- 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java index 60d1719bad..cbfcbbd341 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/FairBlockingQueue.java @@ -138,7 +138,9 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { } /** - * Fair retrieval of an object in the queue. Objects are returned in the order the threads requested them. + * Fair retrieval of an object in the queue. Objects are returned in the order the threads requested them. If the + * wait is interrupted but an object was handed over to this waiter all the same, the object is returned and the + * interrupt status of the calling thread is restored instead of being cleared. * {@inheritDoc} */ @Override @@ -180,9 +182,13 @@ public class FairBlockingQueue<E> implements BlockingQueue<E> { if (null != interruptedException) { // We got interrupted if (null != result) { - // We got a result - clear the interrupt status - // don't propagate cause we have removed a connection from pool - Thread.interrupted(); + // We got a result - deliver it, the connection is not + // lost, but re-assert the interrupt status rather than + // swallow it: a method that returns normally after + // having noticed an interrupt must leave the status + // set, the way the JDK hand-off queues do, so that an + // interrupt meant for a later phase is not lost + Thread.currentThread().interrupt(); } else { throw interruptedException; } diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java index 955caad418..891ddb44f1 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestFairBlockingQueue.java @@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -41,7 +42,9 @@ import org.junit.jupiter.api.Test; * times out, the termination of every outstanding poll once the waiters * are cancelled, the collection operations the queue deliberately * leaves unsupported, the rejection of null elements by every - * insertion path, and the removal contract of the queue iterator. + * insertion path, the removal contract of the queue iterator, and the + * preservation of the interrupt status of a poller that an element is + * handed over to despite its interrupt. */ class TestFairBlockingQueue { @@ -524,6 +527,57 @@ class TestFairBlockingQueue { blocked.join(1000); } + @Test + void deliveredElementDoesNotSwallowTheInterruptOfItsWaiter() throws Exception { + // A waiter interrupted inside poll(timeout) that an element is handed + // over to all the same must receive the element AND keep its interrupt + // status: a normal return after having noticed an interrupt must + // re-assert the status, the way the JDK blocking queues do, or the + // interrupt - often meant for a later phase - is silently lost + final FairBlockingQueue<String> queue = new FairBlockingQueue<>(); + final AtomicReference<String> received = new AtomicReference<>(); + final AtomicReference<Throwable> thrown = new AtomicReference<>(); + final AtomicReference<Boolean> flagAfterPoll = new AtomicReference<>(); + final CountDownLatch done = new CountDownLatch(1); + final Thread waiter = new Thread(() -> { + try { + received.set(queue.poll(30, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + thrown.set(e); + } finally { + flagAfterPoll.set(Thread.currentThread().isInterrupted()); + done.countDown(); + } + }); + waiter.setDaemon(true); + waiter.start(); + awaitUntil(() -> queue.waiters.size() == 1, "the waiter must register on the queue"); + + // Hold the queue lock across the hand-off: after its wait ends with + // the InterruptedException, the waiter blocks in the re-acquisition of + // that lock, so it is pinned between noticing the interrupt and + // deregistering. Throwing the InterruptedException clears the flag, so + // the flag going false again proves the wait has ended with it. + queue.lock.lock(); + try { + waiter.interrupt(); + awaitUntil(() -> !waiter.isInterrupted(), "the waiter must observe its interrupt in the wait"); + Assertions.assertTrue(queue.offer("handed"), + "the element must be handed to the pinned waiter"); + } finally { + queue.lock.unlock(); + } + + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS), + "the waiter must finish once the lock is released"); + Assertions.assertNull(thrown.get(), + "the handed element must not be reported as an interruption, got " + thrown.get()); + Assertions.assertEquals("handed", received.get(), + "the element committed to the waiter's latch must be delivered"); + Assertions.assertEquals(Boolean.TRUE, flagAfterPoll.get(), + "delivering the element must not swallow the waiter's interrupt"); + } + @Test void timedOutAsyncPollRacingAnOfferDoesNotStrandElements() throws Exception { // an offer() that commits its element to a latch concurrently with a @@ -569,4 +623,17 @@ class TestFairBlockingQueue { Assertions.assertEquals(1, queue.size()); Assertions.assertEquals("last", queue.poll()); } + + /* Spins until the condition holds, or fails after a generous deadline. + No sleep decides any outcome: the conditions awaited are the waiter's + own queue registration and the interrupt flag transitions driven by + the wait ending. */ + private static void awaitUntil(BooleanSupplier condition, String message) { + long deadline = System.currentTimeMillis() + 5000; + while (!condition.getAsBoolean()) { + if (System.currentTimeMillis() > deadline) { + Assertions.fail(message); + } + } + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
