This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-5.18.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.18.x by this push:
new a37c111a7 AMQ-9475 - ConsumerControl commands should not auto create
wildcard dests
a37c111a7 is described below
commit a37c111a7522ab0307fc2bba877ce820538f5c3c
Author: Christopher L. Shannon <cshan...@apache.org>
AuthorDate: Wed Apr 10 11:53:12 2024 -0400
AMQ-9475 - ConsumerControl commands should not auto create wildcard
dests
This fixes an issue where wildcard destinations could be inadvertently
created by ConsumerControl commands which could lead to problems. The
processing logic for this command now will only look up existing
destinations if they exist.
---
.../activemq/broker/region/AbstractRegion.java | 26 +++-
.../java/org/apache/activemq/bugs/AMQ9475Test.java | 131 +++++++++++++++++++++
2 files changed, 152 insertions(+), 5 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
index 702b22b15..e62bfed81 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
@@ -21,6 +21,8 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -543,7 +545,11 @@ public abstract class AbstractRegion implements Region {
return sub.pullMessage(context, pull);
}
- protected Destination lookup(ConnectionContext context,
ActiveMQDestination destination,boolean createTemporary) throws Exception {
+ protected Destination lookup(ConnectionContext context,
ActiveMQDestination destination, boolean createTemporary) throws Exception {
+ return lookup(context, destination, createTemporary, true);
+ }
+
+ protected Destination lookup(ConnectionContext context,
ActiveMQDestination destination, boolean createTemporary, boolean autoCreate)
throws Exception {
Destination dest = null;
destinationsLock.readLock().lock();
@@ -553,7 +559,7 @@ public abstract class AbstractRegion implements Region {
destinationsLock.readLock().unlock();
}
- if (dest == null) {
+ if (autoCreate && dest == null) {
if (isAutoCreateDestinations()) {
// Try to auto create the destination... re-invoke broker
// from the
@@ -679,8 +685,8 @@ public abstract class AbstractRegion implements Region {
@Override
public void processConsumerControl(ConsumerBrokerExchange
consumerExchange, ConsumerControl control) {
- Subscription sub = subscriptions.get(control.getConsumerId());
- if (sub != null && sub instanceof AbstractSubscription) {
+ final Subscription sub = subscriptions.get(control.getConsumerId());
+ if (sub instanceof AbstractSubscription) {
((AbstractSubscription)
sub).setPrefetchSize(control.getPrefetch());
if (broker.getDestinationPolicy() != null) {
PolicyEntry entry =
broker.getDestinationPolicy().getEntryFor(control.getDestination());
@@ -691,7 +697,17 @@ public abstract class AbstractRegion implements Region {
LOG.debug("setting prefetch: {}, on subscription: {}; resulting
value: {}",
control.getPrefetch(), control.getConsumerId(),
sub.getConsumerInfo().getPrefetchSize());
try {
- lookup(consumerExchange.getConnectionContext(),
control.getDestination(),false).wakeup();
+ final ActiveMQDestination controlDest =
Objects.requireNonNull(control.getDestination(),
+ "Destination must not be null in ConsumerControl");
+ // Don't auto create patterns (wildcard topics) or composite,
this matches addConsumer()
+ final boolean autoCreate = !controlDest.isPattern() &&
!controlDest.isComposite();
+
+ // If autoCreate is false then lookup() will just return null
if the destination
+ // does not exist and we can skip the call to wakeup. This
will prevent creating
+ // wildcard destinations for wildcard consumers but will use
them if they exist
+
Optional.ofNullable(lookup(consumerExchange.getConnectionContext(),
+ control.getDestination(),false, autoCreate))
+ .ifPresent(Destination::wakeup);
} catch (Exception e) {
LOG.warn("failed to deliver post consumerControl
dispatch-wakeup, to destination: {}", control.getDestination(), e);
}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ9475Test.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ9475Test.java
new file mode 100644
index 000000000..1e39623ac
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ9475Test.java
@@ -0,0 +1,131 @@
+/**
+ * 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.bugs;
+
+import java.util.Arrays;
+import java.util.Collection;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQMessageConsumer;
+import org.apache.activemq.broker.BrokerFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.ConsumerControl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import javax.jms.*;
+import java.net.URI;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+/**
+ * This tests that subscribing to a wildcard and sending a ConsumerControl
+ * command for that wildcard sub will not auto create the destination
+ * by mistake.
+ */
+@RunWith(Parameterized.class)
+public class AMQ9475Test {
+
+ @Parameters(name = "queue={0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { true }, { false } });
+ }
+
+ public AMQ9475Test(boolean queue) {
+ this.destination1 = queue ? new ActiveMQQueue("a.>") : new
ActiveMQTopic("a.>");
+ this.destination2 = queue ? new ActiveMQQueue("a") : new
ActiveMQTopic("a");
+ }
+
+ private BrokerService brokerService;
+ private String connectionUri;
+ private final ActiveMQDestination destination1;
+ private final ActiveMQDestination destination2;
+
+ protected ConnectionFactory createConnectionFactory() throws Exception {
+ ActiveMQConnectionFactory conFactory = new
ActiveMQConnectionFactory(connectionUri);
+ conFactory.setWatchTopicAdvisories(false);
+ return conFactory;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ brokerService = BrokerFactory.createBroker(new
URI("broker://()/localhost?persistent=false&useJmx=true"));
+ brokerService.addConnector("tcp://0.0.0.0:0");
+ brokerService.start();
+ connectionUri =
brokerService.getTransportConnectorByScheme("tcp").getPublishableConnectString();
+ }
+
+ // Normal use case to verify wildcard sub is not created
+ @Test
+ public void testNormalWildcardSub() throws Exception {
+ Session session;
+ try (Connection connection =
createConnectionFactory().createConnection()) {
+ connection.start();
+ session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = session.createConsumer(destination1);
+ sendMessage(session, destination2, "test");
+ assertNotNull(consumer.receive(1000));
+
+
assertNull(brokerService.getBroker().getDestinationMap().get(destination1));
+
assertNotNull(brokerService.getBroker().getDestinationMap().get(destination2));
+ }
+ }
+
+ // Test that the wildcard dest is still not auto-created even after
sending the
+ // ConsumerControl object for it
+ @Test
+ public void testWildcardConsumerControl() throws Exception {
+ Session session;
+ try (ActiveMQConnection connection = (ActiveMQConnection)
createConnectionFactory().createConnection()) {
+ connection.start();
+ session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer)
session.createConsumer(destination1);
+
+ ConsumerControl control = new ConsumerControl();
+ control.setDestination(destination1);
+ control.setConsumerId(consumer.getConsumerId());
+ control.setPrefetch(10);
+ connection.syncSendPacket(control);
+
+ sendMessage(session, destination2, "test");
+ assertNotNull(consumer.receive(1000));
+
+
assertNull(brokerService.getBroker().getDestinationMap().get(destination1));
+
assertNotNull(brokerService.getBroker().getDestinationMap().get(destination2));
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ }
+
+ private void sendMessage(Session session, Destination destination, String
text) throws JMSException {
+ MessageProducer producer = session.createProducer(destination);
+ producer.send(session.createTextMessage(text));
+ producer.close();
+ }
+}
