From d42ee9f4e0802527e8ee73665be0c584428874f9 Mon Sep 17 00:00:00 2001 From: David Ansari Date: Thu, 27 Aug 2026 12:57:30 +0200 Subject: [PATCH] Withhold link credit until the connection is started A consumer created on a connection that has not been started was granted its full prefetch of link credit straight away, in the flow frame that follows its attach. The remote is then free to send, and it does: the messages are transferred on that consumer's link and buffered client side, where the application cannot see them until Connection.start() is called. Jakarta Messaging 6.1.4 says a connection is created in stopped mode, "that means that no messages are being delivered to it", and is emphatic about the guarantee: "clients rely on the fact that no messages will be delivered to a consumer until its connection has been started. Jakarta Messaging Providers must ensure that this is the case." Granting credit before start defeats that. The messages have been delivered to the consumer in every sense the protocol recognises - transferred on its link, and held as unsettled deliveries owned by it at the peer. Only the last hop, into the application, is withheld. That is not merely a semantic quibble, because delivery is exclusive. Once a message has been transferred to one consumer's link it is no longer available to any other consumer on that queue. A consumer on a connection that is never started therefore takes messages that a started consumer on the same queue can never receive. If that connection stays open, the messages are stranded for its lifetime, invisible to every application. This is what the Jakarta Messaging TCK trips over in core20/jmsconsumertests. JmsTool's queue setup creates a connection, a session, a producer and a consumer, and never starts that connection; the test then opens a JMSContext of its own, whose consumer is started, sends one message and blocks in receive(). Both links hold credit, so the peer picks one, and the outcome depends on which - a detail no specification defines. Against RabbitMQ, which selects in attach order, the never-started consumer attached first and always won, so the started consumer blocked forever and three tests hung until killed. Swapping the attach order alone makes the same test pass, which shows how thin the ice is: brokers that select differently pass today with this bug present and unfixed. So grant credit at the point where the specification allows delivery to begin. JmsMessageConsumer.init() now only starts the consumer resource if the session is already started, and JmsSession.start() starts the resource for the consumers that were created while it was stopped. Both paths are needed: JmsSession.start() is once-only and iterates the consumers that exist at the time, so a consumer created after the connection was started - which 6.1.4 explicitly permits - would otherwise never be credited. Between them each consumer is credited exactly once, at the right moment. The pairing of JmsMessageConsumer.start() with startConsumerResource() is the one resumeAfterRollback() already uses for the same purpose. Two existing tests asserted the previous behaviour directly, expecting credit and inbound transfers on a connection that was never started, and checking only that receive() returned null. What they are really about is that the application does not see messages while delivery is paused, which is equally true of a connection that has been stopped, so they now start the connection, let the messages arrive, and stop it before asserting. They are renamed to testNoReceivedMessagesWhenConnectionStopped and testNoReceivedNoWaitMessagesWhenConnectionStopped. Twenty other tests created a consumer on an unstarted connection incidentally, for reasons that have nothing to do with delivery, and simply start the connection now. Connection.stop() has the same shape of problem and is not addressed here: it stops the client side message queue but leaves the link credit in place, so a stopped connection's consumer also goes on taking messages from its queue. Fixing that means draining credit and waiting out the in-flight deliveries, as suspendForRollback() does, and is left for separate work. --- .../apache/qpid/jms/JmsMessageConsumer.java | 11 ++- .../java/org/apache/qpid/jms/JmsSession.java | 3 + .../ConnectionIntegrationTest.java | 2 + .../integration/ConsumerIntegrationTest.java | 72 ++++++++++++++++++- .../integration/SessionIntegrationTest.java | 6 ++ .../SubscriptionsIntegrationTest.java | 8 +++ .../failover/FailoverIntegrationTest.java | 2 + 7 files changed, 101 insertions(+), 3 deletions(-) diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java index 0923f9755..c17b95fc4 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java @@ -141,7 +141,16 @@ public void onPendingFailure(ProviderException cause) { } public void init() throws JMSException { - if (!isPullConsumer()){ + // Link credit is what makes the remote start sending, so it must not be granted while the + // connection is still in stopped mode. A consumer created before Connection.start() gets + // its credit from JmsSession.start() instead. + if (session.isStarted()) { + startConsumerResourceIfNeeded(); + } + } + + void startConsumerResourceIfNeeded() throws JMSException { + if (!isPullConsumer()) { startConsumerResource(); } } diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java index 31d0cd322..572db285d 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java @@ -1179,6 +1179,9 @@ protected void start() throws JMSException { if (started.compareAndSet(false, true)) { for (JmsMessageConsumer consumer : consumers.values()) { consumer.start(); + // Consumers created while the connection was stopped were not granted credit at + // the time, so grant it now that messages may be delivered to them. + consumer.startConsumerResourceIfNeeded(); } } } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java index 7b54c9b87..22a7b64e8 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConnectionIntegrationTest.java @@ -634,6 +634,8 @@ public void testRemotelyEndConnectionWithSessionWithConsumer() throws Exception try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java index 659405635..22e02be60 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java @@ -97,11 +97,67 @@ public class ConsumerIntegrationTest extends QpidJmsTestCase { private final IntegrationTestFixture testFixture = new IntegrationTestFixture(); + @Test + @Timeout(20) + public void testConsumerOnStoppedConnectionIsGrantedNoCreditUntilStarted() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + // Deliberately left in stopped mode. + Connection connection = testFixture.establishConnecton(testPeer); + testPeer.expectBegin(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + + // Attach only. Granting credit here would let the remote send to a connection that has + // not been started, taking messages other consumers on the queue could have had. + testPeer.expectReceiverAttach(); + session.createConsumer(queue); + testPeer.waitForAllHandlersToComplete(1000); + + // Starting the connection is what releases the credit. + testPeer.expectLinkFlow(); + connection.start(); + testPeer.waitForAllHandlersToComplete(1000); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(1000); + } + } + + @Test + @Timeout(20) + public void testConsumerCreatedOnStartedConnectionIsGrantedCreditImmediately() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + Connection connection = testFixture.establishConnecton(testPeer); + connection.start(); + testPeer.expectBegin(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + + // Connection.start() has already happened and is not repeated, so the credit has to + // come from consumer creation itself. + testPeer.expectReceiverAttach(); + testPeer.expectLinkFlow(); + session.createConsumer(queue); + testPeer.waitForAllHandlersToComplete(1000); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(1000); + } + } + @Test @Timeout(20) public void testCloseConsumer() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); testPeer.expectReceiverAttach(); testPeer.expectLinkFlow(); @@ -125,6 +181,8 @@ public void testCloseConsumer() throws Exception { public void testCloseConsumerTimesOut() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setCloseTimeout(500); testPeer.expectBegin(); @@ -160,6 +218,8 @@ public void testRemotelyCloseConsumer() throws Exception { CountDownLatch exceptionFired = new CountDownLatch(1); JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setExceptionListener(new ExceptionListener() { @Override public void onException(JMSException exception) { @@ -628,7 +688,7 @@ public void onMessage(Message m) { @Test @Timeout(20) - public void testNoReceivedMessagesWhenConnectionNotStarted() throws Exception { + public void testNoReceivedMessagesWhenConnectionStopped() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch incoming = new CountDownLatch(1); Connection connection = testFixture.establishConnecton(testPeer); @@ -642,6 +702,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { } }); + // Started, so the consumer is granted credit and the messages reach the client. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -655,6 +717,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { // Wait for a message to arrive then try and receive it, which should not happen // since the connection is not started. assertTrue(incoming.await(10, TimeUnit.SECONDS)); + // Pausing delivery must hide them from the application again. + connection.stop(); assertNull(consumer.receive(100)); testPeer.expectClose(); @@ -666,7 +730,7 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { @Test @Timeout(20) - public void testNoReceivedNoWaitMessagesWhenConnectionNotStarted() throws Exception { + public void testNoReceivedNoWaitMessagesWhenConnectionStopped() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch incoming = new CountDownLatch(1); Connection connection = testFixture.establishConnecton(testPeer); @@ -680,6 +744,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { } }); + // Started, so the consumer is granted credit and the messages reach the client. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -693,6 +759,8 @@ public void onInboundMessage(JmsInboundMessageDispatch envelope) { // Wait for a message to arrive then try and receive it, which should not happen // since the connection is not started. assertTrue(incoming.await(10, TimeUnit.SECONDS)); + // Pausing delivery must hide them from the application again. + connection.stop(); assertNull(consumer.receiveNoWait()); testPeer.expectClose(); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java index c4c06d41d..08ff68512 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java @@ -984,6 +984,8 @@ public void testCreateConsumerSourceContainsTempTopicCapabilityWithoutClientID() private void doCreateConsumerSourceContainsCapabilityTestImpl(Class destType, boolean setClientID) throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer, false, null, null, null, setClientID); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -2000,6 +2002,8 @@ public void testRemotelyEndSessionWithConsumers() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { final CountDownLatch sessionClosed = new CountDownLatch(1); JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.addConnectionListener(new JmsDefaultConnectionListener() { @Override public void onSessionClosed(Session session, Throwable exception) { @@ -2090,6 +2094,8 @@ public boolean isSatisfied() throws Exception { public void testCloseSessionWithConsumerThatRemoteDetaches() throws Exception { try (TestAmqpPeer testPeer = new TestAmqpPeer();) { Connection connection = testFixture.establishConnecton(testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); testPeer.expectBegin(); Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java index 5786ac38f..98548be67 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SubscriptionsIntegrationTest.java @@ -1191,6 +1191,8 @@ public void testRemotelyDetachLinkWithDurableSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch subscriberClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1452,6 +1454,8 @@ public void testRemotelyEndSessionWithDurableSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch sessionClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1537,6 +1541,8 @@ public void testLocallyEndSessionWithSharedConsumer() throws Exception { // Establish connection Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); + // Consumers only get link credit once the connection is started. + connection.start(); final CountDownLatch sessionClosed = new CountDownLatch(1); ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { @@ -1694,6 +1700,8 @@ public void testSharedTopicSubscriberBehavesLikeNoClientIDWasSetWhenAwaitClientI ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:" + serverPort + "?jms.awaitClientID=false"); Connection connection = factory.createConnection(); + // Consumers only get link credit once the connection is started. + connection.start(); // Verify that all handlers complete, i.e. the awaitClientID=false option // setting was effective in provoking the AMQP Open immediately even diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java index 6e176d4cb..d09270de9 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java @@ -2914,6 +2914,8 @@ private void doRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListener testPeer.expectBegin(); final JmsConnection connection = establishAnonymousConnecton("failover.maxReconnectAttempts=1", testPeer); + // Consumers only get link credit once the connection is started. + connection.start(); connection.setExceptionListener(new ExceptionListener() { @Override public void onException(JMSException exception) {