ARTEMIS-5956 Connection.close may leave leaked temporary destinations

Author: clebertsuconic

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

@@ -814,7 +814,7 @@ private void handleTempResource(SimpleString name, boolean queue) {
       // not mean it will get deleted automatically when the session is closed. It is up to the user to delete the
       // resource when finished with it
 
-      TempResourceCleanerUpper cleaner = new TempResourceCleanerUpper(server, name);
+      TempResourceCleanerUpper cleaner = new TempResourceCleanerUpper(server, name, sessionExecutor, context);
       if (remotingConnection instanceof TempResourceObserver observer) {
          cleaner.setObserver(observer);
       }
@@ -1161,24 +1161,42 @@ public String getSecurityDomain() {
       return securityDomain;
    }
 
-   public static class TempResourceCleanerUpper implements CloseListener, FailureListener {
+   public static class TempResourceCleanerUpper implements CloseListener, FailureListener, IOCallback {
+
+      final Executor sessionExecutor;
+
+      final OperationContext operationContext;
 
       private final SimpleString resourceName;
 
       private final ActiveMQServer server;
 
       private TempResourceObserver observer;
 
-      public TempResourceCleanerUpper(final ActiveMQServer server, final SimpleString resourceName) {
+      public TempResourceCleanerUpper(final ActiveMQServer server, final SimpleString resourceName, final Executor sessionExecutor, OperationContext context) {
          this.server = server;
          this.resourceName = resourceName;
+         this.sessionExecutor = sessionExecutor;
+         this.operationContext = context;
       }
 
       public void setObserver(TempResourceObserver observer) {
          this.observer = observer;
       }
 
       private void run() {
+         sessionExecutor.execute(() -> {
+            operationContext.executeOnCompletion(TempResourceCleanerUpper.this);
+         });
+      }
+
+
+      @Override
+      public void onError(int errorCode, String errorMessage) {
+      }
+
+      @Override
+      public void done() {
          try {
             logger.debug("deleting temporary resource {}", resourceName);
             try {
@@ -1197,12 +1215,13 @@ private void run() {
             try {
                AddressInfo a = server.getAddressInfo(resourceName);
                if (a != null && a.isTemporary()) {
-                  server.removeAddressInfo(resourceName, null);
+                  server.removeAddressInfo(resourceName, null, false);
                   if (observer != null) {
                      observer.tempAddressDeleted(resourceName);
                   }
                }
             } catch (ActiveMQException e) {
+               logger.warn(e.getMessage(), e);
                // that's fine.. it can happen due to resource already been deleted
                logger.debug(e.getMessage(), e);
             }

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/TransientQueueManagerImpl.java

@@ -52,7 +52,9 @@ private void doIt() {
    }
 
    public TransientQueueManagerImpl(ActiveMQServer server, SimpleString queueName) {
-      super(server.getExecutorFactory().getExecutor());
+      /* We cannot use a separate executor to delete transient queues as that would
+      *  introduce races on the ServerSessionImpl::TempQueueCleaner. */
+      super();
 
       this.server = server;
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/JMSTemporaryDestinationTest.java

@@ -30,7 +30,9 @@
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.TextMessage;
+import javax.jms.Topic;
 
+import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.tests.util.Wait;
@@ -137,4 +139,30 @@ public void testDeleteTemporaryTopic() throws Exception {
          connection.close();
       }
    }
+
+   @Test
+   public void testAttackDeleteTopicOnClose() throws Exception {
+      // you need to attack open / close many times to reproduce the race I was after when the test was written
+      for (int attack = 0; attack < 100; attack++) {
+         String temporarytopicName;
+
+         try (Connection connection = createConnection()) {
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Topic temporaryTopic = session.createTemporaryTopic();
+
+            MessageProducer producer = session.createProducer(temporaryTopic);
+
+            MessageConsumer consumer = session.createSharedConsumer(temporaryTopic, "mySub");
+            connection.start();
+
+            temporarytopicName = temporaryTopic.getTopicName();
+
+            producer.send(session.createMessage());
+            assertNotNull(consumer.receive(5000));
+         }
+
+         Wait.assertTrue(() -> server.getAddressInfo(SimpleString.of(temporarytopicName)) == null, 5000, 100);
+      }
+   }
+
 }