This closes #1904

artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ThreadLeakCheckRule.java

@@ -230,6 +230,8 @@ public class ThreadLeakCheckRule extends TestWatcher {
 
       if (threadName.contains("SunPKCS11")) {
          return true;
+      } else if (threadName.contains("Keep-Alive-Timer")) {
+         return true;
       } else if (threadName.contains("Attach Listener")) {
          return true;
       } else if ((javaVendor.contains("IBM") || isSystemThread) && threadName.equals("process reaper")) {

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

@@ -438,7 +438,6 @@ public interface ActiveMQServerControl {
    @Attribute(desc = "global maximum limit for in-memory messages, in bytes")
    long getGlobalMaxSize();
 
-
    /**
     * Returns the  memory used by all the addresses on broker for in-memory messages
     */
@@ -466,6 +465,10 @@ public interface ActiveMQServerControl {
    @Operation(desc = "delete an address", impact = MBeanOperationInfo.ACTION)
    void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name) throws Exception;
 
+   @Operation(desc = "delete an address", impact = MBeanOperationInfo.ACTION)
+   void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name,
+                      @Parameter(name = "force", desc = "Force consumers and queues out") boolean force) throws Exception;
+
    /**
     * Create a durable queue.
     * <br>
@@ -481,7 +484,6 @@ public interface ActiveMQServerControl {
    void createQueue(@Parameter(name = "address", desc = "Address of the queue") String address,
                     @Parameter(name = "name", desc = "Name of the queue") String name) throws Exception;
 
-
    /**
     * Create a durable queue.
     * <br>
@@ -498,7 +500,6 @@ public interface ActiveMQServerControl {
                     @Parameter(name = "name", desc = "Name of the queue") String name,
                     @Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception;
 
-
    /**
     * Create a queue.
     * <br>
@@ -572,7 +573,6 @@ public interface ActiveMQServerControl {
                     @Parameter(name = "durable", desc = "Is the queue durable?") boolean durable,
                     @Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception;
 
-
    /**
     * Create a queue.
     * <br>
@@ -616,7 +616,6 @@ public interface ActiveMQServerControl {
                       @Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") Integer maxConsumers,
                       @Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers) throws Exception;
 
-
    /**
     * Update a queue.
     *
@@ -633,7 +632,6 @@ public interface ActiveMQServerControl {
                       @Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers,
                       @Parameter(name = "exclusive", desc = "If the queue should route exclusively to one consumer") Boolean exclusive) throws Exception;
 
-
    /**
     * Deploy a durable queue.
     * <br>
@@ -689,7 +687,6 @@ public interface ActiveMQServerControl {
                      @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;
 
-
    /**
     * Enables message counters for this server.
     */
@@ -987,6 +984,7 @@ public interface ActiveMQServerControl {
                            @Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues,
                            @Parameter(desc = "allow topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics,
                            @Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics) throws Exception;
+
    /**
     * adds a new address setting for a specific address
     */

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

@@ -722,13 +722,19 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
       }
    }
 
+
    @Override
    public void deleteAddress(String name) throws Exception {
+      deleteAddress(name, false);
+   }
+
+   @Override
+   public void deleteAddress(String name, boolean force) throws Exception {
       checkStarted();
 
       clearIO();
       try {
-         server.removeAddressInfo(new SimpleString(name), null);
+         server.removeAddressInfo(new SimpleString(name), null, force);
       } catch (ActiveMQException e) {
          throw new IllegalStateException(e.getMessage());
       } finally {

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java

@@ -533,7 +533,9 @@ public class PageCursorProviderImpl implements PageCursorProvider {
             cursor.confirmPosition(new PagePositionImpl(currentPage.getPageId(), -1));
          }
 
-         while (!storageManager.waitOnOperations(5000)) {
+         // we just need to make sure the storage is done..
+         // if the thread pool is full, we will just log it once instead of looping
+         if (!storageManager.waitOnOperations(5000)) {
             ActiveMQServerLogger.LOGGER.problemCompletingOperations(storageManager.getContext());
          }
       } finally {

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java

@@ -155,7 +155,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
 
    protected BatchingIDGenerator idGenerator;
 
-   protected final ExecutorFactory ioExecutors;
+   protected final ExecutorFactory ioExecutorFactory;
 
    protected final ScheduledExecutorService scheduledExecutorService;
 
@@ -197,15 +197,15 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                                         final CriticalAnalyzer analyzer,
                                         final ExecutorFactory executorFactory,
                                         final ScheduledExecutorService scheduledExecutorService,
-                                        final ExecutorFactory ioExecutors) {
+                                        final ExecutorFactory ioExecutorFactory) {
-      this(config, analyzer, executorFactory, scheduledExecutorService, ioExecutors, null);
+      this(config, analyzer, executorFactory, scheduledExecutorService, ioExecutorFactory, null);
    }
 
    public AbstractJournalStorageManager(Configuration config,
                                         CriticalAnalyzer analyzer,
                                         ExecutorFactory executorFactory,
                                         ScheduledExecutorService scheduledExecutorService,
-                                        ExecutorFactory ioExecutors,
+                                        ExecutorFactory ioExecutorFactory,
                                         IOCriticalErrorListener criticalErrorListener) {
       super(analyzer, CRITICAL_PATHS);
 
@@ -213,7 +213,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
 
       this.ioCriticalErrorListener = criticalErrorListener;
 
-      this.ioExecutors = ioExecutors;
+      this.ioExecutorFactory = ioExecutorFactory;
 
       this.scheduledExecutorService = scheduledExecutorService;
 
@@ -1519,7 +1519,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
 
       beforeStart();
 
-      singleThreadExecutor = executorFactory.getExecutor();
+      singleThreadExecutor = ioExecutorFactory.getExecutor();
 
       bindingsJournal.start();
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java

@@ -129,7 +129,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
       bindingsFF = new NIOSequentialFileFactory(config.getBindingsLocation(), criticalErrorListener, config.getJournalMaxIO_NIO());
       bindingsFF.setDatasync(config.isJournalDatasync());
 
-      Journal localBindings = new JournalImpl(ioExecutors, 1024 * 1024, 2, config.getJournalCompactMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), bindingsFF, "activemq-bindings", "bindings", 1, 0, criticalErrorListener);
+      Journal localBindings = new JournalImpl(ioExecutorFactory, 1024 * 1024, 2, config.getJournalCompactMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), bindingsFF, "activemq-bindings", "bindings", 1, 0, criticalErrorListener);
 
       bindingsJournal = localBindings;
       originalBindingsJournal = localBindings;
@@ -184,7 +184,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
    protected Journal createMessageJournal(Configuration config,
                                         IOCriticalErrorListener criticalErrorListener,
                                         int fileSize) {
-      return new JournalImpl(ioExecutors, fileSize, config.getJournalMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactMinFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), journalFF, "activemq-data", "amq", journalFF.getMaxIO(), 0, criticalErrorListener);
+      return new JournalImpl(ioExecutorFactory, fileSize, config.getJournalMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactMinFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), journalFF, "activemq-data", "amq", journalFF.getMaxIO(), 0, criticalErrorListener);
    }
 
    // Life Cycle Handlers

artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java

@@ -58,6 +58,8 @@ public interface PostOffice extends ActiveMQComponent {
 
    AddressInfo removeAddressInfo(SimpleString address) throws Exception;
 
+   AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception;
+
    AddressInfo getAddressInfo(SimpleString address);
 
    AddressInfo updateAddressInfo(SimpleString addressName, EnumSet<RoutingType> routingTypes) throws Exception;

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

@@ -544,17 +544,32 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
       }
    }
 
+
    @Override
    public AddressInfo removeAddressInfo(SimpleString address) throws Exception {
+      return removeAddressInfo(address, false);
+   }
+
+   @Override
+   public AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception {
       synchronized (addressLock) {
          if (server.hasBrokerPlugins()) {
             server.callBrokerPlugins(plugin -> plugin.beforeRemoveAddress(address));
          }
 
          final Bindings bindingsForAddress = getDirectBindings(address);
+         if (force) {
+            for (Binding binding : bindingsForAddress.getBindings()) {
+               if (binding instanceof QueueBinding) {
+                  ((QueueBinding)binding).getQueue().deleteQueue(true);
+               }
+            }
+
+         } else {
             if (bindingsForAddress.getBindings().size() > 0) {
                throw ActiveMQMessageBundle.BUNDLE.addressHasBindings(address);
             }
+         }
          managementService.unregisterAddress(address);
          final AddressInfo addressInfo = addressManager.removeAddressInfo(address);
          if (server.hasBrokerPlugins()) {

artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java

@@ -262,7 +262,7 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon
             journalLoadInformation[jc.typeByte] = journalsHolder.get(jc).loadSyncOnly(JournalState.SYNCING);
          }
 
-         pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, config.getPagingLocation(), config.getJournalBufferTimeout_NIO(), server.getScheduledPool(), server.getExecutorFactory(), config.isJournalSyncNonTransactional(), criticalErrorListener), server.getAddressSettingsRepository());
+         pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, config.getPagingLocation(), config.getJournalBufferTimeout_NIO(), server.getScheduledPool(), server.getIOExecutorFactory(), config.isJournalSyncNonTransactional(), criticalErrorListener), server.getAddressSettingsRepository());
 
          pageManager.start();
 

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

@@ -511,7 +511,6 @@ public interface ActiveMQServer extends ServiceComponent {
     * @throws Exception
     */
    AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) throws Exception;
-
    /**
     * Remove an {@code AddressInfo} from the broker.
     *
@@ -521,6 +520,16 @@ public interface ActiveMQServer extends ServiceComponent {
     */
    void removeAddressInfo(SimpleString address, SecurityAuth auth) throws Exception;
 
+   /**
+    * Remove an {@code AddressInfo} from the broker.
+    *
+    * @param address the {@code AddressInfo} to remove
+    * @param auth authorization information; {@code null} is valid
+    * @param force It will disconnect everything from the address including queues and consumers
+    * @throws Exception
+    */
+   void removeAddressInfo(SimpleString address, SecurityAuth auth, boolean force) throws Exception;
+
    String getInternalNamingPrefix();
 
 }

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

@@ -2661,14 +2661,20 @@ public class ActiveMQServerImpl implements ActiveMQServer {
       return getAddressInfo(addressInfo.getName());
    }
 
+
    @Override
    public void removeAddressInfo(final SimpleString address, final SecurityAuth auth) throws Exception {
+      removeAddressInfo(address, auth, false);
+   }
+
+   @Override
+   public void removeAddressInfo(final SimpleString address, final SecurityAuth auth, boolean force) throws Exception {
       if (auth != null) {
          securityStore.check(address, CheckType.DELETE_ADDRESS, auth);
       }
 
       AddressInfo addressInfo = getAddressInfo(address);
-      if (postOffice.removeAddressInfo(address) == null) {
+      if (postOffice.removeAddressInfo(address, force) == null) {
          throw ActiveMQMessageBundle.BUNDLE.addressDoesNotExist(address);
       }
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/critical/ShutdownOnCriticalIOErrorMoveNextTest.java

@@ -99,7 +99,7 @@ public class ShutdownOnCriticalIOErrorMoveNextTest extends ActiveMQTestBase {
                protected Journal createMessageJournal(Configuration config,
                                                       IOCriticalErrorListener criticalErrorListener,
                                                       int fileSize) {
-                  return new JournalImpl(ioExecutors, fileSize, config.getJournalMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactMinFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), journalFF, "activemq-data", "amq", journalFF.getMaxIO(), 0, criticalErrorListener) {
+                  return new JournalImpl(ioExecutorFactory, fileSize, config.getJournalMinFiles(), config.getJournalPoolFiles(), config.getJournalCompactMinFiles(), config.getJournalCompactPercentage(), config.getJournalFileOpenTimeout(), journalFF, "activemq-data", "amq", journalFF.getMaxIO(), 0, criticalErrorListener) {
                      @Override
                      protected void moveNextFile(boolean scheduleReclaim) throws Exception {
                         super.moveNextFile(scheduleReclaim);

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java

@@ -150,6 +150,11 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
             proxy.invokeOperation("deleteAddress", name);
          }
 
+         @Override
+         public void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name, @Parameter(name = "force", desc = "Force everything out!") boolean force) throws Exception {
+            proxy.invokeOperation("deleteAddress", name, force);
+         }
+
          @Override
          public void createQueue(final String address,
                                  final String name,

tests/joram-tests/src/test/java/org/apache/activemq/artemis/common/AbstractAdmin.java

@@ -130,7 +130,7 @@ public class AbstractAdmin implements Admin {
    public void deleteQueue(final String name) {
       Boolean result;
       try {
-         invokeSyncOperation(ResourceNames.BROKER, "destroyQueue", name);
+         invokeSyncOperation(ResourceNames.BROKER, "destroyQueue", name, true);
       } catch (Exception e) {
          throw new IllegalStateException(e);
       }
@@ -159,7 +159,7 @@ public class AbstractAdmin implements Admin {
    public void deleteTopic(final String name) {
       Boolean result;
       try {
-         invokeSyncOperation(ResourceNames.BROKER, "deleteAddress", name);
+         invokeSyncOperation(ResourceNames.BROKER, "deleteAddress", name, Boolean.TRUE);
       } catch (Exception e) {
          throw new IllegalStateException(e);
       }

tests/joram-tests/src/test/java/org/objectweb/jtests/jms/framework/PubSubTestCase.java

@@ -149,6 +149,7 @@ public abstract class PubSubTestCase extends JMSTestCase {
          admin.deleteTopicConnectionFactory(PubSubTestCase.TCF_NAME);
          admin.deleteTopic(PubSubTestCase.TOPIC_NAME);
       } catch (Exception ignored) {
+         ignored.printStackTrace();
       } finally {
          publisherTopic = null;
          publisher = null;

tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java

@@ -95,6 +95,11 @@ public class FakePostOffice implements PostOffice {
 
    }
 
+   @Override
+   public AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception {
+      return null;
+   }
+
    @Override
    public boolean addAddressInfo(AddressInfo addressInfo) {
       return false;