ARTEMIS-2030 only use interrupt during shutdown on RA

(cherry picked from commit 99d091a)

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerImpl.java

@@ -409,6 +409,15 @@ public MessageHandler getMessageHandler() throws ActiveMQException {
       return handler;
    }
 
+   @Override
+   public Thread getCurrentThread() {
+      if (onMessageThread != null) {
+         return onMessageThread;
+      }
+      return receiverThread;
+   }
+
+
    // Must be synchronized since messages may be arriving while handler is being set and might otherwise end
    // up not queueing enough executors - so messages get stranded
    @Override

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerInternal.java

@@ -41,6 +41,8 @@ public interface ClientConsumerInternal extends ClientConsumer {
 
    void clear(boolean waitForOnMessage) throws ActiveMQException;
 
+   Thread getCurrentThread();
+
    /**
     * To be called by things like MDBs during shutdown of the server
     *

artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQActivation.java

@@ -26,9 +26,12 @@
 import javax.resource.ResourceException;
 import javax.resource.spi.endpoint.MessageEndpointFactory;
 import javax.resource.spi.work.Work;
+import javax.resource.spi.work.WorkException;
 import javax.resource.spi.work.WorkManager;
 import javax.transaction.xa.XAResource;
 import java.lang.reflect.Method;
+import java.security.AccessController;
+import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -56,6 +59,7 @@
 import org.apache.activemq.artemis.ra.ActiveMQRaUtils;
 import org.apache.activemq.artemis.ra.ActiveMQResourceAdapter;
 import org.apache.activemq.artemis.service.extensions.xa.recovery.XARecoveryConfig;
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
 import org.apache.activemq.artemis.utils.FutureLatch;
 import org.apache.activemq.artemis.utils.PasswordMaskingUtil;
 import org.jboss.logging.Logger;
@@ -242,7 +246,7 @@ public void start() throws ResourceException {
          logger.trace("start()");
       }
       deliveryActive.set(true);
-      ra.getWorkManager().scheduleWork(new SetupActivation());
+      scheduleWork(new SetupActivation());
    }
 
    /**
@@ -282,7 +286,7 @@ public void stop() {
       }
 
       deliveryActive.set(false);
-      teardown();
+      teardown(true);
    }
 
    /**
@@ -348,7 +352,7 @@ protected synchronized void setup() throws Exception {
    /**
     * Teardown the activation
     */
-   protected synchronized void teardown() {
+   protected synchronized void teardown(boolean useInterrupt) {
       logger.debug("Tearing down " + spec);
 
       long timeout = factory == null ? ActiveMQClient.DEFAULT_CALL_TIMEOUT : factory.getCallTimeout();
@@ -369,28 +373,27 @@ protected synchronized void teardown() {
       handlers.clear();
 
       FutureLatch future = new FutureLatch(handlersCopy.length);
-      List<Thread> interruptThreads = new ArrayList<>();
       for (ActiveMQMessageHandler handler : handlersCopy) {
-         Thread thread = handler.interruptConsumer(future);
-         if (thread != null) {
-            interruptThreads.add(thread);
-         }
+         handler.interruptConsumer(future);
       }
 
       //wait for all the consumers to complete any onmessage calls
       boolean stuckThreads = !future.await(timeout);
       //if any are stuck then we need to interrupt them
-      if (stuckThreads) {
-         for (Thread interruptThread : interruptThreads) {
-            try {
-               interruptThread.interrupt();
-            } catch (Exception e) {
-               //ok
+      if (stuckThreads && useInterrupt) {
+         for (ActiveMQMessageHandler handler : handlersCopy) {
+            Thread interruptThread = handler.getCurrentThread();
+            if (interruptThread != null) {
+               try {
+                  interruptThread.interrupt();
+               } catch (Throwable e) {
+                  //ok
+               }
             }
          }
       }
 
-      Thread threadTearDown = new Thread("TearDown/ActiveMQActivation") {
+      Runnable runTearDown = new Runnable() {
          @Override
          public void run() {
             for (ActiveMQMessageHandler handler : handlersCopy) {
@@ -399,10 +402,7 @@ public void run() {
          }
       };
 
-      // We will first start a new thread that will call tearDown on all the instances, trying to graciously shutdown everything.
-      // We will then use the call-timeout to determine a timeout.
-      // if that failed we will then close the connection factory, and interrupt the thread
-      threadTearDown.start();
+      Thread threadTearDown = startThread("TearDown/HornetQActivation", runTearDown);
 
       try {
          threadTearDown.join(timeout);
@@ -550,9 +550,7 @@ protected void setupDestination() throws Exception {
                   calculatedDestinationName = spec.getQueuePrefix() + calculatedDestinationName;
                }
 
-               logger.debug("Unable to retrieve " + destinationName +
-                                                " from JNDI. Creating a new " + destinationType.getName() +
-                                                " named " + calculatedDestinationName + " to be used by the MDB.");
+               logger.debug("Unable to retrieve " + destinationName + " from JNDI. Creating a new " + destinationType.getName() + " named " + calculatedDestinationName + " to be used by the MDB.");
 
                // If there is no binding on naming, we will just create a new instance
                if (isTopic) {
@@ -602,26 +600,49 @@ public String toString() {
       return buffer.toString();
    }
 
-   public void startReconnectThread(final String threadName) {
+   public void startReconnectThread(final String cause) {
       if (logger.isTraceEnabled()) {
-         logger.trace("Starting reconnect Thread " + threadName + " on MDB activation " + this);
+         logger.trace("Starting reconnect Thread " + cause + " on MDB activation " + this);
       }
-      Runnable runnable = new Runnable() {
-         @Override
-         public void run() {
-            reconnect(null);
-         }
-      };
-      Thread t = new Thread(runnable, threadName);
+      try {
+         // We have to use the worker otherwise we may get the wrong classLoader
+         scheduleWork(new ReconnectWork(cause));
+      } catch (Exception e) {
+         logger.warn("Could not reconnect because worker is down", e);
+      }
+   }
+
+   private static Thread startThread(String name, Runnable run) {
+      ClassLoader tccl;
+
+      try {
+         tccl = AccessController.doPrivileged(new PrivilegedExceptionAction<ClassLoader>() {
+            @Override
+            public ClassLoader run() {
+               return ActiveMQActivation.class.getClassLoader();
+            }
+         });
+      } catch (Throwable e) {
+         logger.warn(e.getMessage(), e);
+         tccl = null;
+      }
+
+      ActiveMQThreadFactory factory = new ActiveMQThreadFactory(name, true, tccl);
+      Thread t = factory.newThread(run);
       t.start();
+      return t;
+   }
+
+   private void scheduleWork(Work run) throws WorkException {
+      ra.getWorkManager().scheduleWork(run);
    }
 
    /**
     * Drops all existing connection-related resources and reconnects
     *
     * @param failure if reconnecting in the event of a failure
     */
-   public void reconnect(Throwable failure) {
+   public void reconnect(Throwable failure, boolean useInterrupt) {
       if (logger.isTraceEnabled()) {
          logger.trace("reconnecting activation " + this);
       }
@@ -644,7 +665,7 @@ public void reconnect(Throwable failure) {
       try {
          Throwable lastException = failure;
          while (deliveryActive.get() && (setupAttempts == -1 || reconnectCount < setupAttempts)) {
-            teardown();
+            teardown(useInterrupt);
 
             try {
                Thread.sleep(setupInterval);
@@ -697,7 +718,7 @@ public void run() {
          try {
             setup();
          } catch (Throwable t) {
-            reconnect(t);
+            reconnect(t, false);
          }
       }
 
@@ -706,6 +727,30 @@ public void release() {
       }
    }
 
+   /**
+    * Handles reconnecting
+    */
+   private class ReconnectWork implements Work {
+
+      final String cause;
+
+      ReconnectWork(String cause) {
+         this.cause = cause;
+      }
+
+      @Override
+      public void release() {
+
+      }
+
+      @Override
+      public void run() {
+         logger.tracef("Starting reconnect for %s", cause);
+         reconnect(null, false);
+      }
+
+   }
+
    private class RebalancingListener implements ClusterTopologyListener {
 
       @Override

artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java

@@ -124,17 +124,13 @@ public void setup() throws Exception {
                if (!spec.isShareSubscriptions()) {
                   throw ActiveMQRALogger.LOGGER.canNotCreatedNonSharedSubscriber();
                } else if (ActiveMQRALogger.LOGGER.isDebugEnabled()) {
-                  logger.debug("the mdb on destination " + queueName + " already had " +
-                                                   subResponse.getConsumerCount() +
-                                                   " consumers but the MDB is configured to share subscriptions, so no exceptions are thrown");
+                  logger.debug("the mdb on destination " + queueName + " already had " + subResponse.getConsumerCount() + " consumers but the MDB is configured to share subscriptions, so no exceptions are thrown");
                }
             }
 
             SimpleString oldFilterString = subResponse.getFilterString();
 
-            boolean selectorChanged = selector == null && oldFilterString != null ||
-               oldFilterString == null && selector != null ||
-               (oldFilterString != null && selector != null && !oldFilterString.toString().equals(selector));
+            boolean selectorChanged = selector == null && oldFilterString != null || oldFilterString == null && selector != null || (oldFilterString != null && selector != null && !oldFilterString.toString().equals(selector));
 
             SimpleString oldTopicName = subResponse.getAddress();
 
@@ -198,6 +194,14 @@ XAResource getXAResource() {
       return useXA ? session : null;
    }
 
+   public Thread getCurrentThread() {
+      if (consumer == null) {
+         return null;
+      }
+
+      return consumer.getCurrentThread();
+   }
+
    public Thread interruptConsumer(FutureLatch future) {
       try {
          if (consumer != null) {

tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/client/impl/LargeMessageBufferTest.java

@@ -675,6 +675,11 @@ public ClientMessage receive() throws ActiveMQException {
          return null;
       }
 
+      @Override
+      public Thread getCurrentThread() {
+         return null;
+      }
+
       @Override
       public ClientMessage receive(final long timeout) throws ActiveMQException {
          return null;