activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [1/3] activemq-artemis git commit: ARTEMIS-1324 Deadlock detection and health check of critical components
Date Mon, 07 Aug 2017 22:47:13 GMT
Repository: activemq-artemis
Updated Branches:
  refs/heads/master 8f33d276d -> 01b37de76


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 9269eb3..ecfbf09 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -69,6 +69,7 @@ import org.apache.activemq.artemis.core.server.Consumer;
 import org.apache.activemq.artemis.core.server.HandleStatus;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.QueueFactory;
 import org.apache.activemq.artemis.core.server.RoutingContext;
 import org.apache.activemq.artemis.core.server.ScheduledDeliveryHandler;
 import org.apache.activemq.artemis.core.server.cluster.RemoteQueueBinding;
@@ -92,6 +93,8 @@ import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
 import org.apache.activemq.artemis.utils.collections.PriorityLinkedList;
 import org.apache.activemq.artemis.utils.collections.PriorityLinkedListImpl;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.jboss.logging.Logger;
 
 /**
@@ -99,7 +102,13 @@ import org.jboss.logging.Logger;
  * <p>
  * Completely non blocking between adding to queue and delivering to consumers.
  */
-public class QueueImpl implements Queue {
+public class QueueImpl extends CriticalComponentImpl implements Queue {
+
+   protected static final int CRITICAL_PATHS = 4;
+   protected static final int CRITICAL_PATH_ADD_TAIL = 0;
+   protected static final int CRITICAL_PATH_ADD_HEAD = 1;
+   protected static final int CRITICAL_DELIVER = 2;
+   protected static final int CRITICAL_CONSUMER = 3;
 
    private static final Logger logger = Logger.getLogger(QueueImpl.class);
 
@@ -253,6 +262,8 @@ public class QueueImpl implements Queue {
 
    private volatile RoutingType routingType;
 
+   private final QueueFactory factory;
+
    /**
     * This is to avoid multi-thread races on calculating direct delivery,
     * to guarantee ordering will be always be correct
@@ -332,8 +343,9 @@ public class QueueImpl implements Queue {
                     final StorageManager storageManager,
                     final HierarchicalRepository<AddressSettings> addressSettingsRepository,
                     final Executor executor,
-                    final ActiveMQServer server) {
-      this(id, address, name, filter, null, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server);
+                    final ActiveMQServer server,
+                    final QueueFactory factory) {
+      this(id, address, name, filter, null, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
    }
 
    public QueueImpl(final long id,
@@ -350,8 +362,9 @@ public class QueueImpl implements Queue {
                     final StorageManager storageManager,
                     final HierarchicalRepository<AddressSettings> addressSettingsRepository,
                     final Executor executor,
-                    final ActiveMQServer server) {
-      this(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, RoutingType.MULTICAST, null, null, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server);
+                    final ActiveMQServer server,
+                    final QueueFactory factory) {
+      this(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, RoutingType.MULTICAST, null, null, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
    }
 
    public QueueImpl(final long id,
@@ -371,7 +384,9 @@ public class QueueImpl implements Queue {
                     final StorageManager storageManager,
                     final HierarchicalRepository<AddressSettings> addressSettingsRepository,
                     final Executor executor,
-                    final ActiveMQServer server) {
+                    final ActiveMQServer server,
+                    final QueueFactory factory) {
+      super(server == null ? EmptyCriticalAnalyzer.getInstance() : server.getCriticalAnalyzer(), CRITICAL_PATHS);
 
       this.id = id;
 
@@ -426,6 +441,8 @@ public class QueueImpl implements Queue {
       this.executor = executor;
 
       this.user = user;
+
+      this.factory = factory;
    }
 
    // Bindable implementation -------------------------------------------------------------------------------------
@@ -572,28 +589,42 @@ public class QueueImpl implements Queue {
 
    /* Called when a message is cancelled back into the queue */
    @Override
-   public synchronized void addHead(final MessageReference ref, boolean scheduling) {
-      flushDeliveriesInTransit();
-      if (!scheduling && scheduledDeliveryHandler.checkAndSchedule(ref, false)) {
-         return;
-      }
+   public void addHead(final MessageReference ref, boolean scheduling) {
+      enterCritical(CRITICAL_PATH_ADD_HEAD);
+      synchronized (this) {
+         try {
+            flushDeliveriesInTransit();
+            if (!scheduling && scheduledDeliveryHandler.checkAndSchedule(ref, false)) {
+               return;
+            }
 
-      internalAddHead(ref);
+            internalAddHead(ref);
 
-      directDeliver = false;
+            directDeliver = false;
+         } finally {
+            leaveCritical(CRITICAL_PATH_ADD_HEAD);
+         }
+      }
    }
 
    /* Called when a message is cancelled back into the queue */
    @Override
-   public synchronized void addHead(final List<MessageReference> refs, boolean scheduling) {
-      flushDeliveriesInTransit();
-      for (MessageReference ref : refs) {
-         addHead(ref, scheduling);
-      }
+   public void addHead(final List<MessageReference> refs, boolean scheduling) {
+      enterCritical(CRITICAL_PATH_ADD_HEAD);
+      synchronized (this) {
+         try {
+            flushDeliveriesInTransit();
+            for (MessageReference ref : refs) {
+               addHead(ref, scheduling);
+            }
 
-      resetAllIterators();
+            resetAllIterators();
 
-      deliverAsync();
+            deliverAsync();
+         } finally {
+            leaveCritical(CRITICAL_PATH_ADD_HEAD);
+         }
+      }
    }
 
    @Override
@@ -617,46 +648,46 @@ public class QueueImpl implements Queue {
 
    @Override
    public void addTail(final MessageReference ref, final boolean direct) {
-      if (scheduleIfPossible(ref)) {
-         return;
-      }
+      enterCritical(CRITICAL_PATH_ADD_TAIL);
+      try {
+         if (scheduleIfPossible(ref)) {
+            return;
+         }
 
-      synchronized (directDeliveryGuard) {
-         // The checkDirect flag is periodically set to true, if the delivery is specified as direct then this causes the
-         // directDeliver flag to be re-computed resulting in direct delivery if the queue is empty
-         // We don't recompute it on every delivery since executing isEmpty is expensive for a ConcurrentQueue
-         if (supportsDirectDeliver && !directDeliver &&
-            direct &&
-            System.currentTimeMillis() - lastDirectDeliveryCheck > CHECK_QUEUE_SIZE_PERIOD) {
-            lastDirectDeliveryCheck = System.currentTimeMillis();
-
-            if (intermediateMessageReferences.isEmpty() &&
-               messageReferences.isEmpty() &&
-               !pageIterator.hasNext() &&
-               !pageSubscription.isPaging()) {
-               // We must block on the executor to ensure any async deliveries have completed or we might get out of order
-               // deliveries
-               if (flushExecutor() && flushDeliveriesInTransit()) {
-                  // Go into direct delivery mode
-                  directDeliver = supportsDirectDeliver;
+         synchronized (directDeliveryGuard) {
+            // The checkDirect flag is periodically set to true, if the delivery is specified as direct then this causes the
+            // directDeliver flag to be re-computed resulting in direct delivery if the queue is empty
+            // We don't recompute it on every delivery since executing isEmpty is expensive for a ConcurrentQueue
+            if (supportsDirectDeliver && !directDeliver && direct && System.currentTimeMillis() - lastDirectDeliveryCheck > CHECK_QUEUE_SIZE_PERIOD) {
+               lastDirectDeliveryCheck = System.currentTimeMillis();
+
+               if (intermediateMessageReferences.isEmpty() && messageReferences.isEmpty() && !pageIterator.hasNext() && !pageSubscription.isPaging()) {
+                  // We must block on the executor to ensure any async deliveries have completed or we might get out of order
+                  // deliveries
+                  if (flushExecutor() && flushDeliveriesInTransit()) {
+                     // Go into direct delivery mode
+                     directDeliver = supportsDirectDeliver;
+                  }
                }
             }
          }
-      }
 
-      if (direct && supportsDirectDeliver && directDeliver && deliveriesInTransit.getCount() == 0 && deliverDirect(ref)) {
-         return;
-      }
+         if (direct && supportsDirectDeliver && directDeliver && deliveriesInTransit.getCount() == 0 && deliverDirect(ref)) {
+            return;
+         }
 
-      // We only add queueMemorySize if not being delivered directly
-      queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
+         // We only add queueMemorySize if not being delivered directly
+         queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
 
-      intermediateMessageReferences.add(ref);
+         intermediateMessageReferences.add(ref);
 
-      directDeliver = false;
+         directDeliver = false;
 
-      // Delivery async will both poll for intermediate reference and deliver to clients
-      deliverAsync();
+         // Delivery async will both poll for intermediate reference and deliver to clients
+         deliverAsync();
+      } finally {
+         leaveCritical(CRITICAL_PATH_ADD_TAIL);
+      }
    }
 
    protected boolean scheduleIfPossible(MessageReference ref) {
@@ -788,93 +819,105 @@ public class QueueImpl implements Queue {
          logger.debug(this + " adding consumer " + consumer);
       }
 
-      synchronized (this) {
+      enterCritical(CRITICAL_CONSUMER);
+      try {
+         synchronized (this) {
 
-         if (maxConsumers != MAX_CONSUMERS_UNLIMITED && noConsumers.get() >= maxConsumers) {
-            throw ActiveMQMessageBundle.BUNDLE.maxConsumerLimitReachedForQueue(address, name);
-         }
+            if (maxConsumers != MAX_CONSUMERS_UNLIMITED && noConsumers.get() >= maxConsumers) {
+               throw ActiveMQMessageBundle.BUNDLE.maxConsumerLimitReachedForQueue(address, name);
+            }
 
-         flushDeliveriesInTransit();
+            flushDeliveriesInTransit();
 
-         consumersChanged = true;
+            consumersChanged = true;
 
-         if (!consumer.supportsDirectDelivery()) {
-            this.supportsDirectDeliver = false;
-         }
+            if (!consumer.supportsDirectDelivery()) {
+               this.supportsDirectDeliver = false;
+            }
 
-         cancelRedistributor();
+            cancelRedistributor();
 
-         consumerList.add(new ConsumerHolder(consumer));
+            consumerList.add(new ConsumerHolder(consumer));
 
-         if (consumerSet.add(consumer)) {
-            consumersCount.incrementAndGet();
-         }
+            if (consumerSet.add(consumer)) {
+               consumersCount.incrementAndGet();
+            }
 
-         if (refCountForConsumers != null) {
-            refCountForConsumers.increment();
-         }
+            if (refCountForConsumers != null) {
+               refCountForConsumers.increment();
+            }
 
-         noConsumers.incrementAndGet();
+            noConsumers.incrementAndGet();
+         }
+      } finally {
+         leaveCritical(CRITICAL_CONSUMER);
       }
 
+
    }
 
    @Override
    public void removeConsumer(final Consumer consumer) {
-      synchronized (this) {
-         consumersChanged = true;
 
-         for (ConsumerHolder holder : consumerList) {
-            if (holder.consumer == consumer) {
-               if (holder.iter != null) {
-                  holder.iter.close();
+      enterCritical(CRITICAL_CONSUMER);
+      try {
+         synchronized (this) {
+            consumersChanged = true;
+
+            for (ConsumerHolder holder : consumerList) {
+               if (holder.consumer == consumer) {
+                  if (holder.iter != null) {
+                     holder.iter.close();
+                  }
+                  consumerList.remove(holder);
+                  break;
                }
-               consumerList.remove(holder);
-               break;
             }
-         }
 
-         this.supportsDirectDeliver = checkConsumerDirectDeliver();
+            this.supportsDirectDeliver = checkConsumerDirectDeliver();
 
-         if (pos > 0 && pos >= consumerList.size()) {
-            pos = consumerList.size() - 1;
-         }
+            if (pos > 0 && pos >= consumerList.size()) {
+               pos = consumerList.size() - 1;
+            }
 
-         if (consumerSet.remove(consumer)) {
-            consumersCount.decrementAndGet();
-         }
+            if (consumerSet.remove(consumer)) {
+               consumersCount.decrementAndGet();
+            }
 
-         LinkedList<SimpleString> groupsToRemove = null;
+            LinkedList<SimpleString> groupsToRemove = null;
 
-         for (SimpleString groupID : groups.keySet()) {
-            if (consumer == groups.get(groupID)) {
-               if (groupsToRemove == null) {
-                  groupsToRemove = new LinkedList<>();
+            for (SimpleString groupID : groups.keySet()) {
+               if (consumer == groups.get(groupID)) {
+                  if (groupsToRemove == null) {
+                     groupsToRemove = new LinkedList<>();
+                  }
+                  groupsToRemove.add(groupID);
                }
-               groupsToRemove.add(groupID);
             }
-         }
 
-         // We use an auxiliary List here to avoid concurrent modification exceptions on the keySet
-         // while the iteration is being done.
-         // Since that's a simple HashMap there's no Iterator's support with a remove operation
-         if (groupsToRemove != null) {
-            for (SimpleString groupID : groupsToRemove) {
-               groups.remove(groupID);
+            // We use an auxiliary List here to avoid concurrent modification exceptions on the keySet
+            // while the iteration is being done.
+            // Since that's a simple HashMap there's no Iterator's support with a remove operation
+            if (groupsToRemove != null) {
+               for (SimpleString groupID : groupsToRemove) {
+                  groups.remove(groupID);
+               }
             }
-         }
 
-         if (refCountForConsumers != null) {
-            refCountForConsumers.decrement();
-         }
+            if (refCountForConsumers != null) {
+               refCountForConsumers.decrement();
+            }
 
-         noConsumers.decrementAndGet();
+            noConsumers.decrementAndGet();
+         }
+      } finally {
+         leaveCritical(CRITICAL_CONSUMER);
       }
    }
 
    private boolean checkConsumerDirectDeliver() {
       boolean supports = true;
-      for (ConsumerHolder consumerCheck: consumerList) {
+      for (ConsumerHolder consumerCheck : consumerList) {
          if (!consumerCheck.consumer.supportsDirectDelivery()) {
             supports = false;
          }
@@ -1021,9 +1064,7 @@ public class QueueImpl implements Queue {
       if (pageSubscription != null) {
          // messageReferences will have depaged messages which we need to discount from the counter as they are
          // counted on the pageSubscription as well
-         return messageReferences.size() + getScheduledCount() +
-            deliveringCount.get() +
-            pageSubscription.getMessageCount();
+         return messageReferences.size() + getScheduledCount() + deliveringCount.get() + pageSubscription.getMessageCount();
       } else {
          return messageReferences.size() + getScheduledCount() + deliveringCount.get();
       }
@@ -1505,6 +1546,10 @@ public class QueueImpl implements Queue {
       } catch (Exception e) {
          tx.rollback();
          throw e;
+      } finally {
+         if (factory != null) {
+            factory.queueRemoved(this);
+         }
       }
    }
 
@@ -2286,15 +2331,7 @@ public class QueueImpl implements Queue {
          }
 
          if (logger.isDebugEnabled()) {
-            logger.debug("Queue Memory Size after depage on queue=" + this.getName() +
-                            " is " +
-                            queueMemorySize.get() +
-                            " with maxSize = " +
-                            maxSize +
-                            ". Depaged " +
-                            depaged +
-                            " messages, pendingDelivery=" + messageReferences.size() + ", intermediateMessageReferences= " + intermediateMessageReferences.size() +
-                            ", queueDelivering=" + deliveringCount.get());
+            logger.debug("Queue Memory Size after depage on queue=" + this.getName() + " is " + queueMemorySize.get() + " with maxSize = " + maxSize + ". Depaged " + depaged + " messages, pendingDelivery=" + messageReferences.size() + ", intermediateMessageReferences= " + intermediateMessageReferences.size() + ", queueDelivering=" + deliveringCount.get());
 
          }
       }
@@ -2474,9 +2511,7 @@ public class QueueImpl implements Queue {
       });
    }
 
-   private Pair<String, Binding> locateTargetBinding(SimpleString queueSuffix,
-                                                     Message copyMessage,
-                                                     long oldQueueID) {
+   private Pair<String, Binding> locateTargetBinding(SimpleString queueSuffix, Message copyMessage, long oldQueueID) {
       String targetNodeID = null;
       Binding targetBinding = null;
 
@@ -2528,8 +2563,8 @@ public class QueueImpl implements Queue {
    }
 
    private Message makeCopy(final MessageReference ref,
-                                  final boolean expiry,
-                                  final boolean copyOriginalHeaders) throws Exception {
+                            final boolean expiry,
+                            final boolean copyOriginalHeaders) throws Exception {
       Message message = ref.getMessage();
       /*
        We copy the message and send that to the dla/expiry queue - this is
@@ -2945,8 +2980,13 @@ public class QueueImpl implements Queue {
             // this will avoid that possibility
             // We will be using the deliverRunner instance as the guard object to avoid multiple threads executing
             // an asynchronous delivery
-            synchronized (QueueImpl.this.deliverRunner) {
-               deliver();
+            enterCritical(CRITICAL_DELIVER);
+            try {
+               synchronized (QueueImpl.this.deliverRunner) {
+                  deliver();
+               }
+            } finally {
+               leaveCritical(CRITICAL_DELIVER);
             }
          } catch (Exception e) {
             ActiveMQServerLogger.LOGGER.errorDelivering(e);
@@ -3154,9 +3194,7 @@ public class QueueImpl implements Queue {
       } else {
          if (slowConsumerReaperRunnable == null) {
             scheduleSlowConsumerReaper(settings);
-         } else if (slowConsumerReaperRunnable.checkPeriod != settings.getSlowConsumerCheckPeriod() ||
-            slowConsumerReaperRunnable.threshold != settings.getSlowConsumerThreshold() ||
-            !slowConsumerReaperRunnable.policy.equals(settings.getSlowConsumerPolicy())) {
+         } else if (slowConsumerReaperRunnable.checkPeriod != settings.getSlowConsumerCheckPeriod() || slowConsumerReaperRunnable.threshold != settings.getSlowConsumerThreshold() || !slowConsumerReaperRunnable.policy.equals(settings.getSlowConsumerPolicy())) {
             slowConsumerReaperFuture.cancel(false);
             scheduleSlowConsumerReaper(settings);
          }
@@ -3169,10 +3207,7 @@ public class QueueImpl implements Queue {
       slowConsumerReaperFuture = scheduledExecutor.scheduleWithFixedDelay(slowConsumerReaperRunnable, settings.getSlowConsumerCheckPeriod(), settings.getSlowConsumerCheckPeriod(), TimeUnit.SECONDS);
 
       if (logger.isDebugEnabled()) {
-         logger.debug("Scheduled slow-consumer-reaper thread for queue \"" + getName() +
-                         "\"; slow-consumer-check-period=" + settings.getSlowConsumerCheckPeriod() +
-                         ", slow-consumer-threshold=" + settings.getSlowConsumerThreshold() +
-                         ", slow-consumer-policy=" + settings.getSlowConsumerPolicy());
+         logger.debug("Scheduled slow-consumer-reaper thread for queue \"" + getName() + "\"; slow-consumer-check-period=" + settings.getSlowConsumerCheckPeriod() + ", slow-consumer-threshold=" + settings.getSlowConsumerThreshold() + ", slow-consumer-policy=" + settings.getSlowConsumerPolicy());
       }
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerPlugin.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerPlugin.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerPlugin.java
index b1eab66..b7dcdc3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerPlugin.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerPlugin.java
@@ -38,7 +38,7 @@ import org.apache.activemq.artemis.core.server.impl.AckReason;
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
-
+import org.apache.activemq.artemis.utils.critical.CriticalComponent;
 
 public interface ActiveMQServerPlugin {
 
@@ -399,4 +399,13 @@ public interface ActiveMQServerPlugin {
 
    }
 
+   /**
+    * A Critical failure has been detected.
+    * This will be called before the broker is stopped
+    * @param components
+    * @throws ActiveMQException
+    */
+   default void criticalFailure(CriticalComponent components) throws ActiveMQException {
+   }
+
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/resources/schema/artemis-configuration.xsd
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index fb1d147..03d0896 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -768,6 +768,38 @@
             </xsd:annotation>
          </xsd:element>
 
+         <xsd:element name="critical-analyzer" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  should analyze response time on critical paths and decide for broker shutdown or halt.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
+         <xsd:element name="critical-analyzer-timeout" type="xsd:long" default="120000" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  The default timeout used on analyzing timeouts on the critical path.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
+         <xsd:element name="critical-analyzer-check-period" type="xsd:long" default="0" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  The timeout here will be defaulted to half critical-analyzer-timeout, calculation happening at runtime
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
+         <xsd:element name="critical-analyzer-halt" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  Should the server be shutdown or halted upon critical analysis failure.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
          <xsd:element name="security-settings" maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index ddff9af..bdfa291 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -382,6 +382,11 @@ public class FileConfigurationTest extends ConfigurationImplTest {
       assertEquals(37, conf.getMaxDiskUsage());
       assertEquals(123, conf.getDiskScanPeriod());
 
+      assertEquals(333, conf.getCriticalAnalyzerCheckPeriod());
+      assertEquals(777, conf.getCriticalAnalyzerTimeout());
+      assertEquals(false, conf.isCriticalAnalyzer());
+      assertEquals(true, conf.isCriticalAnalyzerHalt());
+
       assertEquals(false, conf.isJournalDatasync());
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
index 6b5eb9a..ddf702e 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
@@ -51,6 +51,8 @@ import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.ReferenceCounter;
 import org.apache.activemq.artemis.utils.UUID;
 import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
+import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -763,7 +765,7 @@ public class ScheduledDeliveryHandlerTest extends Assert {
       }
    }
 
-   public class FakeQueueForScheduleUnitTest implements Queue {
+   public class FakeQueueForScheduleUnitTest extends CriticalComponentImpl implements Queue {
 
       @Override
       public void setPurgeOnNoConsumers(boolean value) {
@@ -781,6 +783,7 @@ public class ScheduledDeliveryHandlerTest extends Assert {
       }
 
       public FakeQueueForScheduleUnitTest(final int expectedElements) {
+         super(EmptyCriticalAnalyzer.getInstance(), 1);
          this.expectedElements = new CountDownLatch(expectedElements);
       }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
index 6526454..89b7c74 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -57,6 +57,10 @@
       <global-max-size>1234567</global-max-size>
       <max-disk-usage>37</max-disk-usage>
       <disk-scan-period>123</disk-scan-period>
+      <critical-analyzer-halt>true</critical-analyzer-halt>
+      <critical-analyzer-check-period>333</critical-analyzer-check-period>
+      <critical-analyzer-timeout>777</critical-analyzer-timeout>
+      <critical-analyzer>false</critical-analyzer>
       <remoting-incoming-interceptors>
          <class-name>org.apache.activemq.artemis.tests.unit.core.config.impl.TestInterceptor1</class-name>
          <class-name>org.apache.activemq.artemis.tests.unit.core.config.impl.TestInterceptor2</class-name>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/docs/user-manual/en/SUMMARY.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index f87793d..bc9ac31 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -23,6 +23,7 @@
 * [Detecting Dead Connections](connection-ttl.md)
 * [Detecting Slow Consumers](slow-consumers.md)
 * [Avoiding Network Isolation](network-isolation.md)
+* [Detecting Broker Issues (Critical Analysis)](critical-analysis.md)
 * [Resource Manager Configuration](transaction-config.md)
 * [Flow Control](flow-control.md)
 * [Guarantees of sends and commits](send-guarantees.md)

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/docs/user-manual/en/configuration-index.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/configuration-index.md b/docs/user-manual/en/configuration-index.md
index 4c16fa9..c643199 100644
--- a/docs/user-manual/en/configuration-index.md
+++ b/docs/user-manual/en/configuration-index.md
@@ -118,6 +118,11 @@ system-property-prefix | Prefix for replacing configuration settings using Bean
 [network-check-list](network-isolation.md) | The list of pings to be used on ping or InetAddress.isReacheable
 [network-check-ping-command](network-isolation.md) | The command used to oping IPV4 addresses
 [network-check-ping6-command](network-isolation.md) | The command used to oping IPV6 addresses
+[critical-analyzer](critical-analysis.md) | Enable or disable the critical analysis (default true)
+[critical-analyzer-timeout](critical-analysis.md) | Timeout used to do the critical analysis (default 120000 milliseconds)
+[critical-analyzer-check-period](critical-analysis.md) | Time used to check the response times (default half of critical-analyzer-timeout)
+[critical-analyzer-halt](critical-analysis.md) | Should the VM be halted upon failures (default false)
+
 
 #address-setting type
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/docs/user-manual/en/critical-analysis.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/critical-analysis.md b/docs/user-manual/en/critical-analysis.md
new file mode 100644
index 0000000..e867283
--- /dev/null
+++ b/docs/user-manual/en/critical-analysis.md
@@ -0,0 +1,85 @@
+# Critical Analysis of the broker
+
+There are a few things that can go wrong on a production environment:
+
+- Bugs, for more than we try they still happen! We always try to correct them, but that's the only constant in software development.
+- IO Errors, disks and hardware can go bad
+- Memory issues, the CPU can go crazy by another process
+
+For cases like this, we added a protection to the broker to shut itself down when bad things happen.
+
+This is a feature I hope you won't need it, think it as a safeguard:
+
+We measure time response in places like:
+
+- Queue delivery (add to the queue)
+- Journal storage
+- Paging operations
+
+If the response time goes beyond a configured timeout, the broker is considered unstable and an action will be taken to either shutdown the broker or halt the VM.
+
+You can use these following configuration options on broker.xml to configure how the critical analysis is performed.
+
+
+Name | Description
+:--- | :---
+critical-analyzer | Enable or disable the critical analysis (default true)
+critical-analyzer-timeout | Timeout used to do the critical analysis (default 120000 milliseconds)
+critical-analyzer-check-period | Time used to check the response times (default half of critical-analyzer-timeout)
+critical-analyzer-halt | Should the VM be halted upon failures (default false)
+
+The default for critical-analyzer-halt is false, however the generated broker.xml will have it set to true. That is because we cannot halt the VM if you are embedding ActiveMQ Artemis into an application server or on a multi tenant environment.
+
+The broker on the distribution will then have it set to true, but if you use it in any other way the default will be false.
+
+## What would you expect
+
+- You will see some logs
+
+If you have critical-analyzer-halt=true
+
+```
+[Artemis Critical Analyzer] 18:10:00,831 ERROR [org.apache.activemq.artemis.core.server] AMQ224079: The process for the virtual machine will be killed, as component org.apache.activemq.artemis.tests.integration.critical.CriticalSimpleTest$2@5af97850 is not responsive
+```
+
+Or if you have critical-analyzer-halt=false
+
+```
+[Artemis Critical Analyzer] 18:07:53,475 ERROR [org.apache.activemq.artemis.core.server] AMQ224080: The server process will now be stopped, as component org.apache.activemq.artemis.tests.integration.critical.CriticalSimpleTest$2@5af97850 is not responsive
+```
+
+You will see a simple thread dump of the server
+
+```
+[Artemis Critical Analyzer] 18:10:00,836 WARN  [org.apache.activemq.artemis.core.server] AMQ222199: Thread dump: AMQ119001: Generating thread dump
+*******************************************************************************
+===============================================================================
+AMQ119002: Thread Thread[Thread-1 (ActiveMQ-scheduled-threads),5,main] name = Thread-1 (ActiveMQ-scheduled-threads) id = 19 group = java.lang.ThreadGroup[name=main,maxpri=10]
+
+sun.misc.Unsafe.park(Native Method)
+java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
+java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
+java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
+java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
+java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
+java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
+java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
+java.lang.Thread.run(Thread.java:745)
+===============================================================================
+
+
+..... blablablablaba ..........
+
+
+===============================================================================
+AMQ119003: End Thread dump
+*******************************************************************************
+
+```
+
+- The Server will be halted if configured to halt
+
+- The system will be stopped if no halt is used:
+* Notice that if the system is not behaving well, there is no guarantees the stop will work.
+
+

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
index 0de19a1..afe1403 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
@@ -239,7 +239,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
                              final Executor executor, final ActiveMQServer server) {
             super(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, deliveryMode,
                   maxConsumers, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager,
-                  addressSettingsRepository, executor, server);
+                  addressSettingsRepository, executor, server, null);
          }
 
          @Override
@@ -375,7 +375,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
       // Forcing a situation where the server would unexpectedly create a duplicated queue. The server should still start normally
       LocalQueueBinding newBinding = new LocalQueueBinding(QUEUE,
                                                            new QueueImpl(queueID, QUEUE, QUEUE, null, null, true, false,
-                                                                         false, null, null, null, null, null, null),
+                                                                         false, null, null, null, null, null, null, null),
                                                            server.getNodeID());
       server.getStorageManager().addQueueBinding(txID, newBinding);
       server.getStorageManager().commitBindings(txID);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/InterruptedLargeMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/InterruptedLargeMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/InterruptedLargeMessageTest.java
index f926af1..66381fa 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/InterruptedLargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/InterruptedLargeMessageTest.java
@@ -520,7 +520,7 @@ public class InterruptedLargeMessageTest extends LargeMessageTestBase {
                         HierarchicalRepository<AddressSettings> addressSettingsRepository,
                         Executor executor) {
             super(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, scheduledExecutor,
-                  postOffice, storageManager, addressSettingsRepository, executor, null);
+                  postOffice, storageManager, addressSettingsRepository, executor, null, null);
          }
 
          @Override

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/critical/CriticalSimpleTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/critical/CriticalSimpleTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/critical/CriticalSimpleTest.java
new file mode 100644
index 0000000..da25243
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/critical/CriticalSimpleTest.java
@@ -0,0 +1,102 @@
+/**
+ * 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.artemis.tests.integration.critical;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerPlugin;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.utils.critical.CriticalComponent;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CriticalSimpleTest extends ActiveMQTestBase {
+   @Test
+   public void testSimpleShutdown() throws Exception {
+
+      Configuration configuration = createDefaultConfig(false);
+      configuration.setCriticalAnalyzerCheckPeriod(10).setCriticalAnalyzerHalt(true);
+      ActiveMQServer server = createServer(false, configuration, AddressSettings.DEFAULT_PAGE_SIZE, AddressSettings.DEFAULT_MAX_SIZE_BYTES);
+      server.start();
+
+      try {
+
+         CountDownLatch latch = new CountDownLatch(1);
+
+         server.getConfiguration().registerBrokerPlugin(new ActiveMQServerPlugin() {
+            @Override
+            public void criticalFailure(CriticalComponent components) throws ActiveMQException {
+               latch.countDown();
+            }
+         });
+
+         server.getCriticalAnalyzer().add(new CriticalComponent() {
+            @Override
+            public boolean isExpired(long timeout) {
+               return true;
+            }
+         });
+
+         Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+         Wait.waitFor(() -> !server.isStarted());
+
+
+         Assert.assertFalse(server.isStarted());
+
+      } finally {
+         server.stop();
+      }
+
+
+   }
+
+   @Test
+   public void testCriticalOff() throws Exception {
+
+      Configuration configuration = createDefaultConfig(false);
+      configuration.setCriticalAnalyzerCheckPeriod(10).setCriticalAnalyzer(false);
+      ActiveMQServer server = createServer(false, configuration, AddressSettings.DEFAULT_PAGE_SIZE, AddressSettings.DEFAULT_MAX_SIZE_BYTES);
+      server.start();
+
+      try {
+         server.getCriticalAnalyzer().add(new CriticalComponent() {
+            @Override
+            public boolean isExpired(long timeout) {
+               return true;
+            }
+         });
+
+         Wait.waitFor(() -> !server.isStarted(), 500, 10);
+
+
+         Assert.assertTrue(server.isStarted());
+
+      } finally {
+         server.stop();
+      }
+
+
+   }
+
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
index 532207e..bae9115 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
@@ -87,7 +87,7 @@ public class TopicCleanupTest extends JMSTestBase {
                                               FilterImpl.createFilter(Filter.GENERIC_IGNORED_FILTER), null,
                                               true, false, false, server.getScheduledPool(), server.getPostOffice(),
                                               storage, server.getAddressSettingsRepository(),
-                                              server.getExecutorFactory().getExecutor(), server);
+                                              server.getExecutorFactory().getExecutor(), server, null);
 
             LocalQueueBinding binding = new LocalQueueBinding(queue.getAddress(), queue, server.getNodeID());
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java
index c830459..b97a861 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java
@@ -54,6 +54,7 @@ import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
 import org.apache.activemq.artemis.utils.IDGenerator;
 import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
 import org.apache.activemq.artemis.utils.SimpleIDGenerator;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.jboss.logging.Logger;
 import org.junit.After;
 import org.junit.Assert;
@@ -1724,7 +1725,7 @@ public class NIOJournalCompactTest extends JournalImplTestBase {
 
       final ExecutorService deleteExecutor = Executors.newCachedThreadPool(ActiveMQThreadFactory.defaultThreadFactory());
 
-      final JournalStorageManager storage = new JournalStorageManager(config, factory, iofactory);
+      final JournalStorageManager storage = new JournalStorageManager(config, EmptyCriticalAnalyzer.getInstance(), factory, iofactory);
 
       storage.start();
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/DeleteMessagesOnStartupTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/DeleteMessagesOnStartupTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/DeleteMessagesOnStartupTest.java
index 615a924..4819c5d 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/DeleteMessagesOnStartupTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/DeleteMessagesOnStartupTest.java
@@ -34,6 +34,7 @@ import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.impl.PostOfficeJournalLoader;
 import org.apache.activemq.artemis.tests.unit.core.postoffice.impl.FakeQueue;
 import org.apache.activemq.artemis.tests.unit.core.server.impl.fakes.FakePostOffice;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runners.Parameterized;
@@ -90,7 +91,7 @@ public class DeleteMessagesOnStartupTest extends StorageManagerTestBase {
 
    @Override
    protected JournalStorageManager createJournalStorageManager(Configuration configuration) {
-      return new JournalStorageManager(configuration, execFactory, execFactory) {
+      return new JournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), execFactory, execFactory) {
          @Override
          public void deleteMessage(final long messageID) throws Exception {
             deletedMessage.add(messageID);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/JournalFileSizeTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/JournalFileSizeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/JournalFileSizeTest.java
index 749715b..f25e950 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/JournalFileSizeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/JournalFileSizeTest.java
@@ -1,4 +1,5 @@
-/** * Licensed to the Apache Software Foundation (ASF) under one or more
+/**
+ * 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
@@ -15,13 +16,13 @@
  */
 package org.apache.activemq.artemis.tests.integration.persistence;
 
-
 import java.io.File;
 
 import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
 import org.apache.activemq.artemis.core.io.aio.AIOSequentialFileFactory;
 import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
 import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -43,9 +44,7 @@ public class JournalFileSizeTest {
       ConfigurationImpl config = new ConfigurationImpl();
       int origFileSize = config.getJournalFileSize();
       config.setJournalFileSize(origFileSize + (align / 2 - 1));
-      JournalStorageManager manager = new JournalStorageManager(config,
-            new OrderedExecutorFactory(null),
-            new OrderedExecutorFactory(null));
+      JournalStorageManager manager = new JournalStorageManager(config, EmptyCriticalAnalyzer.getInstance(), new OrderedExecutorFactory(null), new OrderedExecutorFactory(null));
       int fileSize = manager.getMessageJournal().getFileSize();
       Assert.assertEquals(origFileSize, fileSize);
    }
@@ -55,9 +54,7 @@ public class JournalFileSizeTest {
       ConfigurationImpl config = new ConfigurationImpl();
       int origFileSize = config.getJournalFileSize();
       config.setJournalFileSize(origFileSize + (align / 2 + 1));
-      JournalStorageManager manager = new JournalStorageManager(config,
-            new OrderedExecutorFactory(null),
-            new OrderedExecutorFactory(null));
+      JournalStorageManager manager = new JournalStorageManager(config, EmptyCriticalAnalyzer.getInstance(), new OrderedExecutorFactory(null), new OrderedExecutorFactory(null));
       int fileSize = manager.getMessageJournal().getFileSize();
       Assert.assertEquals(origFileSize + align, fileSize);
    }
@@ -67,9 +64,7 @@ public class JournalFileSizeTest {
       ConfigurationImpl config = new ConfigurationImpl();
       int origFileSize = config.getJournalFileSize();
       config.setJournalFileSize(origFileSize + (align / 2));
-      JournalStorageManager manager = new JournalStorageManager(config,
-            new OrderedExecutorFactory(null),
-            new OrderedExecutorFactory(null));
+      JournalStorageManager manager = new JournalStorageManager(config,EmptyCriticalAnalyzer.getInstance(), new OrderedExecutorFactory(null), new OrderedExecutorFactory(null));
       int fileSize = manager.getMessageJournal().getFileSize();
       Assert.assertEquals(origFileSize + align, fileSize);
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/RestartSMTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/RestartSMTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/RestartSMTest.java
index 2ee879f..48bdc9a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/RestartSMTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/RestartSMTest.java
@@ -31,6 +31,7 @@ import org.apache.activemq.artemis.tests.unit.core.server.impl.fakes.FakeJournal
 import org.apache.activemq.artemis.tests.unit.core.server.impl.fakes.FakePostOffice;
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -66,7 +67,7 @@ public class RestartSMTest extends ActiveMQTestBase {
 
       PostOffice postOffice = new FakePostOffice();
 
-      final JournalStorageManager journal = new JournalStorageManager(createDefaultInVMConfig(), execFactory, execFactory);
+      final JournalStorageManager journal = new JournalStorageManager(createDefaultInVMConfig(), EmptyCriticalAnalyzer.getInstance(), execFactory, execFactory);
 
       try {
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/StorageManagerTestBase.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/StorageManagerTestBase.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/StorageManagerTestBase.java
index 508f23b..6479d1e 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/StorageManagerTestBase.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/StorageManagerTestBase.java
@@ -38,6 +38,7 @@ import org.apache.activemq.artemis.tests.unit.core.server.impl.fakes.FakePostOff
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.utils.ExecutorFactory;
 import org.apache.activemq.artemis.utils.TimeAndCounterIDGenerator;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.runner.RunWith;
@@ -138,7 +139,7 @@ public abstract class StorageManagerTestBase extends ActiveMQTestBase {
     * @param configuration
     */
    protected JournalStorageManager createJournalStorageManager(Configuration configuration) {
-      JournalStorageManager jsm = new JournalStorageManager(configuration, execFactory, execFactory);
+      JournalStorageManager jsm = new JournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), execFactory, execFactory);
       addActiveMQComponent(jsm);
       return jsm;
    }
@@ -147,7 +148,7 @@ public abstract class StorageManagerTestBase extends ActiveMQTestBase {
     * @param configuration
     */
    protected JDBCJournalStorageManager createJDBCJournalStorageManager(Configuration configuration) {
-      JDBCJournalStorageManager jsm = new JDBCJournalStorageManager(configuration, execFactory, execFactory, scheduledExecutorService);
+      JDBCJournalStorageManager jsm = new JDBCJournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), execFactory, execFactory, scheduledExecutorService);
       addActiveMQComponent(jsm);
       return jsm;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
index 3095ab9..8236702 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
@@ -87,6 +87,7 @@ import org.apache.activemq.artemis.tests.util.TransportConfigurationUtils;
 import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
 import org.apache.activemq.artemis.utils.ExecutorFactory;
 import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -435,7 +436,7 @@ public final class ReplicationTest extends ActiveMQTestBase {
     * @throws Exception
     */
    private JournalStorageManager getStorage() throws Exception {
-      return new JournalStorageManager(createDefaultInVMConfig(), factory, factory);
+      return new JournalStorageManager(createDefaultInVMConfig(), EmptyCriticalAnalyzer.getInstance(), factory, factory);
    }
 
    /**

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
----------------------------------------------------------------------
diff --git a/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java b/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
index ed1a4e5..b5dc463 100644
--- a/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
+++ b/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
@@ -70,7 +70,7 @@ public class QueueImplTest extends ActiveMQTestBase {
       QueueImpl queue =
                new QueueImpl(1, new SimpleString("address1"), new SimpleString("queue1"), null, null, false, true,
                              false, scheduledExecutor, null, null, null,
-                             Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null);
+                             Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null, null);
 
       // Send one scheduled
 
@@ -135,7 +135,7 @@ public class QueueImplTest extends ActiveMQTestBase {
 
    @Test
    public void testScheduled() throws Exception {
-      QueueImpl queue = new QueueImpl(1, new SimpleString("address1"), new SimpleString("queue1"), null, null, false, true, false, scheduledExecutor, null, null, null, Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null);
+      QueueImpl queue = new QueueImpl(1, new SimpleString("address1"), new SimpleString("queue1"), null, null, false, true, false, scheduledExecutor, null, null, null, Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null, null);
 
       FakeConsumer consumer = null;
 
@@ -233,7 +233,7 @@ public class QueueImplTest extends ActiveMQTestBase {
          public void disconnect() {
          }
       };
-      QueueImpl queue = new QueueImpl(1, new SimpleString("address1"), QueueImplTest.queue1, null, null, false, true, false, scheduledExecutor, null, null, null, Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null);
+      QueueImpl queue = new QueueImpl(1, new SimpleString("address1"), QueueImplTest.queue1, null, null, false, true, false, scheduledExecutor, null, null, null, Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory()), null, null);
       MessageReference messageReference = generateReference(queue, 1);
       queue.addConsumer(consumer);
       messageReference.setScheduledDeliveryTime(System.currentTimeMillis() + 2000);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java
index 21bc48d..985e83f 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java
@@ -42,6 +42,7 @@ import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
 import org.apache.activemq.artemis.utils.ExecutorFactory;
 import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
 import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -93,7 +94,7 @@ public class DuplicateDetectionUnitTest extends ActiveMQTestBase {
 
          ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(ActiveMQDefaultConfiguration.getDefaultScheduledThreadPoolMaxSize(), ActiveMQThreadFactory.defaultThreadFactory());
 
-         journal = new JournalStorageManager(configuration, factory, factory);
+         journal = new JournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), factory, factory);
 
          journal.start();
          journal.loadBindingJournal(new ArrayList<QueueBindingInfo>(), new ArrayList<GroupingInfo>(), new ArrayList<AddressBindingInfo>());
@@ -113,7 +114,7 @@ public class DuplicateDetectionUnitTest extends ActiveMQTestBase {
 
          journal.stop();
 
-         journal = new JournalStorageManager(configuration, factory, factory);
+         journal = new JournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), factory, factory);
          journal.start();
          journal.loadBindingJournal(new ArrayList<QueueBindingInfo>(), new ArrayList<GroupingInfo>(), new ArrayList<AddressBindingInfo>());
 
@@ -136,7 +137,7 @@ public class DuplicateDetectionUnitTest extends ActiveMQTestBase {
 
          mapDups.clear();
 
-         journal = new JournalStorageManager(configuration, factory, factory);
+         journal = new JournalStorageManager(configuration, EmptyCriticalAnalyzer.getInstance(), factory, factory);
          journal.start();
          journal.loadBindingJournal(new ArrayList<QueueBindingInfo>(), new ArrayList<GroupingInfo>(), new ArrayList<AddressBindingInfo>());
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
index efb9e77..54cae7b 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
@@ -35,8 +35,10 @@ import org.apache.activemq.artemis.core.server.impl.AckReason;
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.utils.ReferenceCounter;
 import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
+import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
 
-public class FakeQueue implements Queue {
+public class FakeQueue extends CriticalComponentImpl implements Queue {
 
    @Override
    public void setPurgeOnNoConsumers(boolean value) {
@@ -175,6 +177,7 @@ public class FakeQueue implements Queue {
    }
 
    public FakeQueue(final SimpleString name, final long id) {
+      super(EmptyCriticalAnalyzer.getInstance(), 1);
       this.name = name;
       this.id = id;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/QueueImplTest.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/QueueImplTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/QueueImplTest.java
index b64ff03..40f9214 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/QueueImplTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/QueueImplTest.java
@@ -1310,6 +1310,6 @@ public class QueueImplTest extends ActiveMQTestBase {
 
    private QueueImpl getQueue(SimpleString name, boolean durable, boolean temporary, Filter filter) {
       return new QueueImpl(1, QueueImplTest.address1, name, filter, null, durable, temporary, false, scheduledExecutor,
-                           new FakePostOffice(), null, null, executor, null);
+                           new FakePostOffice(), null, null, executor, null, null);
    }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeQueueFactory.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeQueueFactory.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeQueueFactory.java
index 40c117a..4721579 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeQueueFactory.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeQueueFactory.java
@@ -42,7 +42,7 @@ public final class FakeQueueFactory implements QueueFactory {
    public Queue createQueueWith(final QueueConfig config) {
       return new QueueImpl(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(),
                            config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(),
-                           scheduledExecutor, postOffice, null, null, executor, null);
+                           scheduledExecutor, postOffice, null, null, executor, null, this);
    }
 
    @Deprecated
@@ -57,7 +57,7 @@ public final class FakeQueueFactory implements QueueFactory {
                             final boolean temporary,
                             final boolean autoCreated) {
       return new QueueImpl(persistenceID, address, name, filter, subscription, user, durable, temporary, autoCreated,
-                           scheduledExecutor, postOffice, null, null, executor, null);
+                           scheduledExecutor, postOffice, null, null, executor, null, this);
    }
 
    @Override


Mime
View raw message