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

This closes #3833

This commit is contained in:
Justin Bertram 2021-11-02 12:59:46 -05:00
parent 7988739f41 1d0c0a8897
commit e41ec90a66
10 changed files with 92 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
View File

@ -1090,7 +1090,7 @@ public interface ActiveMQServerControl {
@Operation(desc = "Destroy a queue", impact = MBeanOperationInfo.ACTION)
void destroyQueue(@Parameter(name = "name", desc = "Name of the queue to destroy") String name,
@Parameter(name = "removeConsumers", desc = "Remove consumers of this queue") boolean removeConsumers,
@Parameter(name = "autoDeleteAddress", desc = "Automatically delete the address if this was the last queue") boolean autoDeleteAddress) throws Exception;
@Parameter(name = "forceAutoDeleteAddress", desc = "Automatically delete the address if this was the last queue") boolean autoDeleteAddress) throws Exception;
/**
* Enables message counters for this server.

2
artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/server/impl/JMSServerManagerImpl.java
View File

@ -817,7 +817,7 @@ public class JMSServerManagerImpl extends CleaningActivateCallback implements JM
// We can't remove the remote binding. As this would be the bridge associated with the topic on this case
if (binding.getType() != BindingType.REMOTE_QUEUE) {
server.destroyQueue(SimpleString.toSimpleString(queueName), null, !removeConsumers, removeConsumers, true);
server.destroyQueue(SimpleString.toSimpleString(queueName), null, !removeConsumers, removeConsumers, false);
}
}

6
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
View File

@ -1563,9 +1563,9 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
}
@Override
public void destroyQueue(final String name, final boolean removeConsumers, final boolean autoDeleteAddress) throws Exception {
public void destroyQueue(final String name, final boolean removeConsumers, final boolean forceAutoDeleteAddress) throws Exception {
if (AuditLogger.isBaseLoggingEnabled()) {
AuditLogger.destroyQueue(this.server, null, null, name, removeConsumers, autoDeleteAddress);
AuditLogger.destroyQueue(this.server, null, null, name, removeConsumers, forceAutoDeleteAddress);
}
checkStarted();
@ -1573,7 +1573,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
try {
SimpleString queueName = new SimpleString(name);
try {
server.destroyQueue(queueName, null, !removeConsumers, removeConsumers, autoDeleteAddress);
server.destroyQueue(queueName, null, !removeConsumers, removeConsumers, forceAutoDeleteAddress);
} catch (Exception e) {
if (AuditLogger.isResourceLoggingEnabled()) {
AuditLogger.destroyQueueFailure(name);

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
View File

@ -672,13 +672,13 @@ public interface ActiveMQServer extends ServiceComponent {
SecurityAuth session,
boolean checkConsumerCount,
boolean removeConsumers,
boolean autoDeleteAddress) throws Exception;
boolean forceAutoDeleteAddress) throws Exception;
void destroyQueue(SimpleString queueName,
SecurityAuth session,
boolean checkConsumerCount,
boolean removeConsumers,
boolean autoDeleteAddress,
boolean forceAutoDeleteAddress,
boolean checkMessageCount) throws Exception;
String destroyConnectionWithSessionMetadata(String metaKey, String metaValue) throws Exception;

18
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
View File

@ -2329,9 +2329,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
throw ActiveMQMessageBundle.BUNDLE.noSuchQueue(queueName);
}
String address = binding.getAddress().toString();
destroyQueue(queueName, session, checkConsumerCount, removeConsumers, addressSettingsRepository.getMatch(address).isAutoDeleteAddresses());
destroyQueue(queueName, session, checkConsumerCount, removeConsumers, false);
}
@Override
@ -2339,8 +2337,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final SecurityAuth session,
final boolean checkConsumerCount,
final boolean removeConsumers,
final boolean autoDeleteAddress) throws Exception {
destroyQueue(queueName, session, checkConsumerCount, removeConsumers, autoDeleteAddress, false);
final boolean forceAutoDeleteAddress) throws Exception {
destroyQueue(queueName, session, checkConsumerCount, removeConsumers, forceAutoDeleteAddress, false);
}
@Override
@ -2348,7 +2346,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final SecurityAuth session,
final boolean checkConsumerCount,
final boolean removeConsumers,
final boolean autoDeleteAddress,
final boolean forceAutoDeleteAddress,
final boolean checkMessageCount) throws Exception {
if (postOffice == null) {
return;
@ -2384,7 +2382,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
if (hasBrokerQueuePlugins()) {
callBrokerQueuePlugins(plugin -> plugin.beforeDestroyQueue(queue, session, checkConsumerCount, removeConsumers, autoDeleteAddress));
callBrokerQueuePlugins(plugin -> plugin.beforeDestroyQueue(queue, session, checkConsumerCount, removeConsumers, forceAutoDeleteAddress));
}
if (mirrorControllerService != null) {
@ -2394,13 +2392,13 @@ public class ActiveMQServerImpl implements ActiveMQServer {
queue.deleteQueue(removeConsumers);
if (hasBrokerQueuePlugins()) {
callBrokerQueuePlugins(plugin -> plugin.afterDestroyQueue(queue, address, session, checkConsumerCount, removeConsumers, autoDeleteAddress));
callBrokerQueuePlugins(plugin -> plugin.afterDestroyQueue(queue, address, session, checkConsumerCount, removeConsumers, forceAutoDeleteAddress));
}
if (queue.isTemporary()) {
if (forceAutoDeleteAddress) {
AddressInfo addressInfo = getAddressInfo(address);
if (autoDeleteAddress && postOffice != null && addressInfo != null && addressInfo.isAutoCreated() && !isAddressBound(address.toString()) && addressSettingsRepository.getMatch(address.toString()).getAutoDeleteAddressesDelay() == 0) {
if (postOffice != null && addressInfo != null && addressInfo.isAutoCreated() && !isAddressBound(address.toString()) && addressSettingsRepository.getMatch(address.toString()).getAutoDeleteAddressesDelay() == 0) {
try {
removeAddressInfo(address, session);
} catch (ActiveMQDeleteAddressException e) {

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueManagerImpl.java
View File

@ -72,7 +72,7 @@ public class QueueManagerImpl extends ReferenceCounterUtil implements QueueManag
ActiveMQServerLogger.LOGGER.autoRemoveQueue("" + queue.getName(), queue.getID(), "" + queue.getAddress());
try {
server.destroyQueue(queueName, null, true, false, settings.isAutoDeleteAddresses(), true);
server.destroyQueue(queueName, null, true, false, false, true);
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.errorRemovingAutoCreatedDestination(e, queueName, "queue");
}

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
View File

@ -1129,7 +1129,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
logger.debug("deleting temporary queue " + bindingName);
}
try {
server.destroyQueue(bindingName, null, false);
server.destroyQueue(bindingName, null, false, false, true);
if (observer != null) {
observer.tempQueueDeleted(bindingName);
}
@ -1177,7 +1177,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
throw new ActiveMQNonExistentQueueException();
}
server.destroyQueue(unPrefixedQueueName, this, true);
server.destroyQueue(unPrefixedQueueName, this, true, false, true);
TempQueueCleanerUpper cleaner = this.tempQueueCleannerUppers.remove(unPrefixedQueueName);

6
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/TopicDurableTests.java
View File

@ -39,6 +39,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.Test;
@ -47,6 +48,11 @@ import static org.hamcrest.CoreMatchers.is;
public class TopicDurableTests extends JMSClientTestSupport {
@Override
protected void addConfiguration(ActiveMQServer server) {
server.getConfiguration().setAddressQueueScanPeriod(100);
}
@Test
public void testMessageDurableSubscription() throws Exception {
JmsConnectionFactory connectionFactory = new JmsConnectionFactory(getBrokerQpidJMSConnectionURI() + "?jms.clientID=jmsTopicClient");

38
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TemporaryDestinationTest.java
View File

@ -17,6 +17,7 @@
package org.apache.activemq.artemis.tests.integration.jms.client;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
@ -35,7 +36,9 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.client.ActiveMQConnection;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.Before;
import org.junit.Test;
@ -278,6 +281,8 @@ public class TemporaryDestinationTest extends JMSTestBase {
for (ServerSession serverSession : server.getSessions()) {
assertFalse(((ServerSessionImpl)serverSession).cloneTargetAddresses().containsKey(SimpleString.toSimpleString(temporaryQueue.getQueueName())));
}
Wait.assertTrue(() -> server.locateQueue(temporaryQueue.getQueueName()) == null, 1000, 100);
Wait.assertTrue(() -> server.getAddressInfo(SimpleString.toSimpleString(temporaryQueue.getQueueName())) == null, 1000, 100);
} finally {
if (conn != null) {
conn.close();
@ -285,6 +290,39 @@ public class TemporaryDestinationTest extends JMSTestBase {
}
}
@Test
public void testTemporaryQueueConnectionClosedRemovedAMQP() throws Exception {
testTemporaryQueueConnectionClosedRemoved("AMQP");
}
@Test
public void testTemporaryQueueConnectionClosedRemovedCORE() throws Exception {
testTemporaryQueueConnectionClosedRemoved("CORE");
}
@Test
public void testTemporaryQueueConnectionClosedRemovedOpenWire() throws Exception {
testTemporaryQueueConnectionClosedRemoved("OPENWIRE");
}
private void testTemporaryQueueConnectionClosedRemoved(String protocol) throws Exception {
ConnectionFactory factory = CFUtil.createConnectionFactory("amqp", "tcp://localhost:61616");
final TemporaryQueue temporaryQueue;
try (Connection conn = factory.createConnection()) {
Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
temporaryQueue = s.createTemporaryQueue();
MessageProducer producer = s.createProducer(temporaryQueue);
producer.send(s.createMessage());
// These next two assertions are here to validate the test itself
// The queue and address should be found on the server while they still exist on the connection
Wait.assertFalse(() -> server.locateQueue(temporaryQueue.getQueueName()) == null, 1000, 100);
Wait.assertFalse(() -> server.getAddressInfo(SimpleString.toSimpleString(temporaryQueue.getQueueName())) == null, 1000, 100);
}
Wait.assertTrue(() -> server.locateQueue(temporaryQueue.getQueueName()) == null, 1000, 100);
Wait.assertTrue(() -> server.getAddressInfo(SimpleString.toSimpleString(temporaryQueue.getQueueName())) == null, 1000, 100);
}
@Test
public void testForTempQueueTargetInfosSizeLimit() throws Exception {
try {

30
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
View File

@ -81,6 +81,7 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.DeletionPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
@ -506,6 +507,35 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
checkNoResource(ObjectNameBuilder.DEFAULT.getAddressObjectName(address));
}
@Test
public void testCreateAndDestroyQueueWithAutoDeleteAddress() throws Exception {
server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoDeleteAddresses(false));
SimpleString address = RandomUtil.randomSimpleString();
SimpleString name = RandomUtil.randomSimpleString();
ActiveMQServerControl serverControl = createManagementControl();
checkNoResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(address, name, RoutingType.ANYCAST));
if (legacyCreateQueue) {
serverControl.createQueue(address.toString(), "ANYCAST", name.toString(), null, true, -1, false, true);
} else {
serverControl.createQueue(new QueueConfiguration(name).setAddress(address).setRoutingType(RoutingType.ANYCAST).setAutoCreateAddress(true).toJSON());
}
checkResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(address, name, RoutingType.ANYCAST));
QueueControl queueControl = ManagementControlHelper.createQueueControl(address, name, RoutingType.ANYCAST, mbeanServer);
Assert.assertEquals(address.toString(), queueControl.getAddress());
Assert.assertEquals(name.toString(), queueControl.getName());
Assert.assertNull(queueControl.getFilter());
Assert.assertEquals(true, queueControl.isDurable());
Assert.assertEquals(false, queueControl.isTemporary());
serverControl.destroyQueue(name.toString(), false, true);
checkNoResource(ObjectNameBuilder.DEFAULT.getQueueObjectName(address, name, RoutingType.ANYCAST));
checkNoResource(ObjectNameBuilder.DEFAULT.getAddressObjectName(address));
}
@Test
public void testRemoveQueueFilter() throws Exception {