Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Christopher L. Shannon 2024-04-10 11:53:12 -04:00
parent fdc4a63476
commit a37c111a75
2 changed files with 152 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
View File

@ -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);
}

131
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ9475Test.java Normal file
View File

@ -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();
}
}