diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java index 900938ec04c..c5670716a72 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java @@ -147,6 +147,8 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon * This strictly rejects non-standard property types such as Character, Map, and List. */ private boolean strictCompliance = false; + private boolean deferPrefetchUntilStarted = false; + private final AtomicBoolean everStarted = new AtomicBoolean(false); private boolean disableTimeStampsByDefault; private boolean optimizedMessageDispatch = true; @@ -579,6 +581,7 @@ public void setClientInternalExceptionListener(ClientInternalExceptionListener l public void start() throws JMSException { checkClosedOrFailed(); ensureConnectionInfoSent(); + everStarted.set(true); if (started.compareAndSet(false, true)) { for (Iterator i = sessions.iterator(); i.hasNext();) { ActiveMQSession session = i.next(); @@ -1057,6 +1060,25 @@ public void setStrictCompliance(boolean strictCompliance) { this.strictCompliance = strictCompliance; } + public boolean isDeferPrefetchUntilStarted() { + return deferPrefetchUntilStarted; + } + + /** + * See {@link ActiveMQConnectionFactory#setDeferPrefetchUntilStarted(boolean)}. + */ + public void setDeferPrefetchUntilStarted(boolean deferPrefetchUntilStarted) { + this.deferPrefetchUntilStarted = deferPrefetchUntilStarted; + } + + /** + * @return true once {@link #start()} has been called at least once, + * regardless of later {@link #stop()} calls. + */ + public boolean isEverStarted() { + return everStarted.get(); + } + public boolean isExclusiveConsumer() { return exclusiveConsumer; } diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java index 93cfb20d309..81c965ef404 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java @@ -131,6 +131,7 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne * This strictly rejects non-standard property types such as Character, Map, and List. */ private boolean strictCompliance = false; + private boolean deferPrefetchUntilStarted = false; private boolean disableTimeStampsByDefault; private boolean optimizedMessageDispatch = true; @@ -428,6 +429,7 @@ protected ActiveMQConnection createActiveMQConnection(Transport transport, JMSSt protected void configureConnection(ActiveMQConnection connection) throws JMSException { connection.setPrefetchPolicy(getPrefetchPolicy()); + connection.setDeferPrefetchUntilStarted(isDeferPrefetchUntilStarted()); connection.setDisableTimeStampsByDefault(isDisableTimeStampsByDefault()); connection.setOptimizedMessageDispatch(isOptimizedMessageDispatch()); connection.setCopyMessageOnSend(isCopyMessageOnSend()); @@ -1059,6 +1061,22 @@ public void setStrictCompliance(boolean strictCompliance) { this.strictCompliance = strictCompliance; } + public boolean isDeferPrefetchUntilStarted() { + return deferPrefetchUntilStarted; + } + + /** + * When enabled, queue consumers created before their connection is first + * started register with a prefetch of zero so the broker does not dispatch + * messages into a consumer that cannot deliver them. The configured + * prefetch is restored when the connection starts. Default is false to + * preserve the historical behavior of buffering pre-start dispatches in + * the client. + */ + public void setDeferPrefetchUntilStarted(boolean deferPrefetchUntilStarted) { + this.deferPrefetchUntilStarted = deferPrefetchUntilStarted; + } + public String getClientIDPrefix() { return clientIDPrefix; } diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java index a6bf1b20952..e30af11c4a6 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java @@ -44,6 +44,7 @@ import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.command.ActiveMQObjectMessage; import org.apache.activemq.command.ActiveMQTempDestination; +import org.apache.activemq.command.ConsumerControl; import org.apache.activemq.command.CommandTypes; import org.apache.activemq.command.ConsumerId; import org.apache.activemq.command.ConsumerInfo; @@ -154,6 +155,9 @@ class PreviouslyDelivered { private final String selector; private boolean synchronizationRegistered; private final AtomicBoolean started = new AtomicBoolean(false); + // Configured prefetch withheld until the connection first starts; zero when + // no deferral is pending. See ActiveMQConnectionFactory#setDeferPrefetchUntilStarted. + private volatile int deferredPrefetchSize; private MessageAvailableListener availableListener; @@ -293,6 +297,21 @@ public ActiveMQMessageConsumer(ActiveMQSession session, ConsumerId consumerId, A || this.nonBlockingRedelivery || session.connection.isMessagePrioritySupported(); this.consumerExpiryCheckEnabled = session.connection.isConsumerExpiryCheckEnabled(); + // Register queue consumers created before the connection has ever been + // started with zero prefetch so the broker does not push messages into a + // consumer that cannot deliver them; such messages would park in the held + // dispatch channel and starve running consumers. The configured prefetch + // is restored when the connection starts. + if (this.session.connection.isDeferPrefetchUntilStarted() + && !this.session.connection.isEverStarted() + && dest.isQueue() + && !browser + && this.info.getPrefetchSize() > 0) { + this.deferredPrefetchSize = this.info.getPrefetchSize(); + this.info.setPrefetchSize(0); + this.info.setCurrentPrefetchSize(0); + } + if (messageListener != null) { setMessageListener(messageListener); } @@ -448,7 +467,7 @@ public MessageListener getMessageListener() throws JMSException { @Override public void setMessageListener(MessageListener listener) throws JMSException { checkClosed(); - if (info.getPrefetchSize() == 0) { + if (info.getPrefetchSize() == 0 && deferredPrefetchSize == 0) { throw new JMSException("Illegal prefetch size of zero. This setting is not supported for asynchronous consumers please set a value of at least 1"); } if (listener != null) { @@ -665,7 +684,7 @@ public Message receive(long timeout) throws JMSException { while (timeout > 0) { MessageDispatch md; - if (info.getPrefetchSize() == 0) { + if (isPullConsumer()) { md = dequeue(-1); // We let the broker let us know when we timeout. } else { md = dequeue(timeout); @@ -697,7 +716,7 @@ public Message receiveNoWait() throws JMSException { sendPullCommand(-1); MessageDispatch md; - if (info.getPrefetchSize() == 0) { + if (isPullConsumer()) { md = dequeue(-1); // We let the broker let us know when we // timeout. } else { @@ -914,7 +933,9 @@ protected void checkClosed() throws IllegalStateException { */ protected void sendPullCommand(long timeout) throws JMSException { clearDeliveredList(); - if (info.getCurrentPrefetchSize() == 0 && unconsumedMessages.isEmpty()) { + // A consumer whose prefetch is deferred until connection start is not a + // pull consumer; a pull here could steal a message into the held channel. + if (info.getCurrentPrefetchSize() == 0 && deferredPrefetchSize == 0 && unconsumedMessages.isEmpty()) { MessagePull messagePull = new MessagePull(); messagePull.configure(info); messagePull.setTimeout(timeout); @@ -1485,7 +1506,7 @@ public void dispatch(MessageDispatch md) { // Pull consumer needs to check if pull timed out and send // a new pull command if not. - if (info.getCurrentPrefetchSize() == 0) { + if (info.getCurrentPrefetchSize() == 0 && deferredPrefetchSize == 0) { unconsumedMessages.enqueue(null); } } @@ -1628,11 +1649,41 @@ public void start() throws JMSException { if (unconsumedMessages.isClosed()) { return; } + restoreDeferredPrefetch(); started.set(true); unconsumedMessages.start(); session.executor.wakeup(); } + /** + * A pull consumer relies on the broker to answer each MessagePull, including + * signalling the receive timeout. A consumer whose prefetch is merely deferred + * until the connection starts carries prefetch zero on the wire but sends no + * pulls, so it must use client-side timeouts like any push consumer. + */ + private boolean isPullConsumer() { + return info.getPrefetchSize() == 0 && deferredPrefetchSize == 0; + } + + /** + * Restores the configured prefetch that was withheld while the connection + * had never been started and tells the broker so it re-credits this + * consumer and dispatches any pending messages. + */ + private void restoreDeferredPrefetch() throws JMSException { + var prefetch = deferredPrefetchSize; + if (prefetch > 0) { + deferredPrefetchSize = 0; + info.setPrefetchSize(prefetch); + info.setCurrentPrefetchSize(prefetch); + var control = new ConsumerControl(); + control.setConsumerId(info.getConsumerId()); + control.setDestination(info.getDestination()); + control.setPrefetch(prefetch); + session.asyncSendPacket(control); + } + } + public void stop() { started.set(false); unconsumedMessages.stop(); diff --git a/activemq-tooling/activemq-jakarta-messaging-tck/src/main/java/org/apache/activemq/tck/JNDIInitialContextFactory.java b/activemq-tooling/activemq-jakarta-messaging-tck/src/main/java/org/apache/activemq/tck/JNDIInitialContextFactory.java index a95a86eeb23..1e060cbe9f5 100644 --- a/activemq-tooling/activemq-jakarta-messaging-tck/src/main/java/org/apache/activemq/tck/JNDIInitialContextFactory.java +++ b/activemq-tooling/activemq-jakarta-messaging-tck/src/main/java/org/apache/activemq/tck/JNDIInitialContextFactory.java @@ -97,6 +97,11 @@ public Context getInitialContext(final Hashtable environment) throws Namin private static ActiveMQConnectionFactory createConnectionFactory(final String clientId) { final ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL); factory.setNestedMapAndListEnabled(false); + // JmsTool creates competing queue consumers on connections it never starts. + // Without deferral the broker round-robins messages into those consumers' + // prefetch, where they park in the held dispatch channel and the started + // consumer's receive() blocks forever (core20 jmsconsumertests queueReceiveTests). + factory.setDeferPrefetchUntilStarted(true); if (clientId != null) { factory.setClientID(clientId); } diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/UnstartedConnectionQueueDispatchTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/UnstartedConnectionQueueDispatchTest.java new file mode 100644 index 00000000000..4f73ac71709 --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/UnstartedConnectionQueueDispatchTest.java @@ -0,0 +1,170 @@ +/** + * 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.activemq; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import jakarta.jms.Connection; +import jakarta.jms.Session; + +import org.apache.activemq.broker.BrokerService; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * A queue consumer whose connection has never been started must not attract + * push dispatch. Historically the broker round-robins messages into such a + * consumer's prefetch, where they sit in the client's held dispatch channel + * and starve the consumers that are actually running (seen as the TCK + * queueReceiveTests hang). + */ +public class UnstartedConnectionQueueDispatchTest { + + private static final int MESSAGE_COUNT = 10; + + private BrokerService broker; + private String connectionUri; + private Connection startedConnection; + private Connection unstartedConnection; + + @Before + public void setUp() throws Exception { + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.setAdvisorySupport(false); + broker.setSchedulerSupport(false); + broker.addConnector("vm://localhost"); + broker.start(); + broker.waitUntilStarted(); + connectionUri = "vm://localhost"; + } + + @After + public void tearDown() throws Exception { + if (startedConnection != null) { + try { startedConnection.close(); } catch (Exception ignored) {} + } + if (unstartedConnection != null) { + try { unstartedConnection.close(); } catch (Exception ignored) {} + } + if (broker != null) { + broker.stop(); + broker.waitUntilStopped(); + } + } + + private ActiveMQConnectionFactory createFactory() { + var factory = new ActiveMQConnectionFactory(connectionUri); + factory.setDeferPrefetchUntilStarted(true); + return factory; + } + + @Test(timeout = 60000) + public void testStartedConsumerReceivesAllMessagesDespiteUnstartedCompetitor() throws Exception { + var factory = createFactory(); + + // competing consumer on a connection that is never started + unstartedConnection = factory.createConnection(); + var unstartedSession = unstartedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var queue = unstartedSession.createQueue("test.unstarted.dispatch"); + var neverStarted = unstartedSession.createConsumer(queue); + assertNotNull(neverStarted); + + // active consumer on a started connection + startedConnection = factory.createConnection(); + startedConnection.start(); + var session = startedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var active = session.createConsumer(queue); + + var producer = session.createProducer(queue); + for (int i = 0; i < MESSAGE_COUNT; i++) { + producer.send(session.createTextMessage("message-" + i)); + } + + // every message must reach the running consumer; none may park in the + // never-started consumer's prefetch + for (int i = 0; i < MESSAGE_COUNT; i++) { + var received = active.receive(5000); + assertNotNull("Message " + i + " was dispatched to the never-started consumer", received); + } + } + + @Test(timeout = 60000) + public void testDeferredConsumerReceivesMessageReleasedByClosedConsumer() throws Exception { + // TCK core/queueConnection connNotStartedQueueTest shape: a started + // receiver prefetches two messages and consumes one; closing it returns + // the other to the queue; a receiver on a never-started connection must + // see nothing until start, then receive the released message. + var factory = createFactory(); + + startedConnection = factory.createConnection(); + startedConnection.start(); + var session = startedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var queue = session.createQueue("test.unstarted.released"); + var first = session.createConsumer(queue); + var producer = session.createProducer(queue); + producer.send(session.createTextMessage("one")); + producer.send(session.createTextMessage("two")); + assertNotNull(first.receive(5000)); + first.close(); + + unstartedConnection = factory.createConnection(); + var unstartedSession = unstartedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var second = unstartedSession.createConsumer(queue); + assertNull("No delivery before the connection is started", second.receive(1000)); + + unstartedConnection.start(); + assertNotNull("Released message must be delivered once the connection starts", second.receive(5000)); + } + + @Test(timeout = 60000) + public void testDeferredConsumerReceivesAfterConnectionStart() throws Exception { + var factory = createFactory(); + + unstartedConnection = factory.createConnection(); + var session = unstartedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var queue = session.createQueue("test.unstarted.recovery"); + var consumer = session.createConsumer(queue); + + // the async pattern: listener registered before the connection starts + final var delivered = new CountDownLatch(MESSAGE_COUNT); + consumer.setMessageListener(message -> delivered.countDown()); + + startedConnection = factory.createConnection(); + startedConnection.start(); + var producerSession = startedConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + var producer = producerSession.createProducer(queue); + for (int i = 0; i < MESSAGE_COUNT; i++) { + producer.send(producerSession.createTextMessage("message-" + i)); + } + + // nothing may be delivered while the connection is not started + assertTrue("Messages must not be delivered before start", delivered.getCount() == MESSAGE_COUNT); + + // starting the connection restores the prefetch credit and delivery flows + unstartedConnection.start(); + assertTrue("Messages should be delivered after connection start", + delivered.await(10, TimeUnit.SECONDS)); + } +}