You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2017/08/07 22:47:14 UTC
[2/3] activemq-artemis git commit: ARTEMIS-1324 Deadlock detection
and health check of critical components
ARTEMIS-1324 Deadlock detection and health check of critical components
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/f16af753
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/f16af753
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/f16af753
Branch: refs/heads/master
Commit: f16af7535417e3ba3ad625d22430edda7b76b5d2
Parents: 8f33d27
Author: Clebert Suconic <cl...@apache.org>
Authored: Sat Aug 5 00:52:42 2017 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Aug 7 18:40:03 2017 -0400
----------------------------------------------------------------------
.../cli/commands/tools/xml/XmlDataExporter.java | 3 +-
.../artemis/cli/commands/etc/broker.xml | 9 +
.../apache/activemq/cli/test/ArtemisTest.java | 5 +-
.../artemis/utils/critical/CriticalAction.java | 22 ++
.../utils/critical/CriticalAnalyzer.java | 43 +++
.../utils/critical/CriticalAnalyzerImpl.java | 185 ++++++++++++
.../utils/critical/CriticalComponent.java | 54 ++++
.../utils/critical/CriticalComponentImpl.java | 68 +++++
.../artemis/utils/critical/CriticalMeasure.java | 52 ++++
.../utils/critical/EmptyCriticalAnalyzer.java | 90 ++++++
.../utils/critical/CriticalAnalyzerTest.java | 114 +++++++
.../config/ActiveMQDefaultConfiguration.java | 26 ++
.../artemis/core/config/Configuration.java | 17 ++
.../core/config/impl/ConfigurationImpl.java | 55 ++++
.../deployers/impl/FileConfigurationParser.java | 8 +
.../journal/AbstractJournalStorageManager.java | 15 +-
.../impl/journal/JDBCJournalStorageManager.java | 7 +-
.../impl/journal/JournalStorageManager.java | 14 +-
.../artemis/core/server/ActiveMQServer.java | 3 +
.../core/server/ActiveMQServerLogger.java | 8 +
.../activemq/artemis/core/server/Queue.java | 3 +-
.../artemis/core/server/QueueFactory.java | 4 +
.../core/server/impl/ActiveMQServerImpl.java | 94 +++++-
.../core/server/impl/LastValueQueue.java | 6 +-
.../core/server/impl/QueueFactoryImpl.java | 17 +-
.../artemis/core/server/impl/QueueImpl.java | 295 +++++++++++--------
.../server/plugin/ActiveMQServerPlugin.java | 11 +-
.../resources/schema/artemis-configuration.xsd | 32 ++
.../core/config/impl/FileConfigurationTest.java | 5 +
.../impl/ScheduledDeliveryHandlerTest.java | 5 +-
.../resources/ConfigurationTest-full-config.xml | 4 +
docs/user-manual/en/SUMMARY.md | 1 +
docs/user-manual/en/configuration-index.md | 5 +
docs/user-manual/en/critical-analysis.md | 85 ++++++
.../integration/client/HangConsumerTest.java | 4 +-
.../client/InterruptedLargeMessageTest.java | 2 +-
.../critical/CriticalSimpleTest.java | 102 +++++++
.../jms/client/TopicCleanupTest.java | 2 +-
.../journal/NIOJournalCompactTest.java | 3 +-
.../DeleteMessagesOnStartupTest.java | 3 +-
.../persistence/JournalFileSizeTest.java | 17 +-
.../integration/persistence/RestartSMTest.java | 3 +-
.../persistence/StorageManagerTestBase.java | 5 +-
.../replication/ReplicationTest.java | 3 +-
.../timing/core/server/impl/QueueImplTest.java | 6 +-
.../impl/DuplicateDetectionUnitTest.java | 7 +-
.../unit/core/postoffice/impl/FakeQueue.java | 5 +-
.../unit/core/server/impl/QueueImplTest.java | 2 +-
.../server/impl/fakes/FakeQueueFactory.java | 4 +-
49 files changed, 1349 insertions(+), 184 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/xml/XmlDataExporter.java
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/xml/XmlDataExporter.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/xml/XmlDataExporter.java
index f54c3d4..d37c965 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/xml/XmlDataExporter.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/xml/XmlDataExporter.java
@@ -83,6 +83,7 @@ import org.apache.activemq.artemis.core.settings.impl.HierarchicalObjectReposito
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;
@Command(name = "exp", description = "Export all message-data using an XML that could be interpreted by any system.")
public final class XmlDataExporter extends OptionalLocking {
@@ -134,7 +135,7 @@ public final class XmlDataExporter extends OptionalLocking {
final ExecutorService executor = Executors.newFixedThreadPool(5, ActiveMQThreadFactory.defaultThreadFactory());
ExecutorFactory executorFactory = new OrderedExecutorFactory(executor);
- storageManager = new JournalStorageManager(config, executorFactory, executorFactory);
+ storageManager = new JournalStorageManager(config, EmptyCriticalAnalyzer.getInstance(), executorFactory, executorFactory);
XMLOutputFactory factory = XMLOutputFactory.newInstance();
XMLStreamWriter rawXmlWriter = factory.createXMLStreamWriter(out, "UTF-8");
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
index 603938a..adabf85 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
@@ -59,6 +59,15 @@ ${ping-config.settings}${journal-buffer.settings}${connector-config.settings}
that won't support flow control. -->
<max-disk-usage>90</max-disk-usage>
+ <!-- should the broker detect dead locks and other issues -->
+ <critical-analyzer>true</critical-analyzer>
+
+ <critical-analyzer-timeout>120000</critical-analyzer-timeout>
+
+ <critical-analyzer-check-period>60000</critical-analyzer-check-period>
+
+ <critical-analyzer-halt>true</critical-analyzer-halt>
+
${global-max-section}
<acceptors>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-cli/src/test/java/org/apache/activemq/cli/test/ArtemisTest.java
----------------------------------------------------------------------
diff --git a/artemis-cli/src/test/java/org/apache/activemq/cli/test/ArtemisTest.java b/artemis-cli/src/test/java/org/apache/activemq/cli/test/ArtemisTest.java
index 803020c..8b4098a 100644
--- a/artemis-cli/src/test/java/org/apache/activemq/cli/test/ArtemisTest.java
+++ b/artemis-cli/src/test/java/org/apache/activemq/cli/test/ArtemisTest.java
@@ -541,10 +541,11 @@ public class ArtemisTest extends CliTestBase {
Artemis.main("create", instanceFolder.getAbsolutePath(), "--force", "--silent", "--no-web", "--queues", queues, "--addresses", addresses, "--no-autotune", "--require-login");
System.setProperty("artemis.instance", instanceFolder.getAbsolutePath());
- // Some exceptions may happen on the initialization, but they should be ok on start the basic core protocol
- Artemis.internalExecute("run");
try {
+ // Some exceptions may happen on the initialization, but they should be ok on start the basic core protocol
+ Artemis.internalExecute("run");
+
try (ServerLocator locator = ServerLocatorImpl.newLocator("tcp://localhost:61616");
ClientSessionFactory factory = locator.createSessionFactory();
ClientSession coreSession = factory.createSession("admin", "admin", false, true, true, false, 0)) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAction.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAction.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAction.java
new file mode 100644
index 0000000..bd16d46
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAction.java
@@ -0,0 +1,22 @@
+/**
+ * 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.utils.critical;
+
+public interface CriticalAction {
+
+ void run(CriticalComponent failedComponent);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java
new file mode 100644
index 0000000..6b5a436
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java
@@ -0,0 +1,43 @@
+/**
+ * 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.utils.critical;
+
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+
+public interface CriticalAnalyzer extends ActiveMQComponent {
+
+ default void clear() {
+ }
+
+ boolean isMeasuring();
+
+ void add(CriticalComponent component);
+
+ void remove(CriticalComponent component);
+
+ CriticalAnalyzer setCheckTime(long timeout);
+
+ long getCheckTime();
+
+ CriticalAnalyzer setTimeout(long timeout);
+
+ long getTimeout();
+
+ CriticalAnalyzer addAction(CriticalAction action);
+
+ void check();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
new file mode 100644
index 0000000..c583f2a
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
@@ -0,0 +1,185 @@
+/**
+ * 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.utils.critical;
+
+import java.util.ConcurrentModificationException;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
+import org.jboss.logging.Logger;
+
+public class CriticalAnalyzerImpl implements CriticalAnalyzer {
+
+ private final Logger logger = Logger.getLogger(CriticalAnalyzer.class);
+
+ private volatile long timeout;
+
+ private volatile long checkTime;
+
+ @Override
+ public void clear() {
+ actions.clear();
+ components.clear();
+ }
+
+ private CopyOnWriteArrayList<CriticalAction> actions = new CopyOnWriteArrayList<>();
+
+ private Thread thread;
+
+ private final Semaphore running = new Semaphore(1);
+
+ private final ConcurrentHashSet<CriticalComponent> components = new ConcurrentHashSet<>();
+
+ @Override
+ public boolean isMeasuring() {
+ return true;
+ }
+
+ @Override
+ public void add(CriticalComponent component) {
+ components.add(component);
+ }
+
+ @Override
+ public void remove(CriticalComponent component) {
+ components.remove(component);
+ }
+
+ @Override
+ public CriticalAnalyzer setCheckTime(long timeout) {
+ this.checkTime = timeout;
+ return this;
+ }
+
+ @Override
+ public long getCheckTime() {
+ if (checkTime == 0) {
+ checkTime = getTimeout() / 2;
+ }
+ return checkTime;
+ }
+
+ @Override
+ public CriticalAnalyzer setTimeout(long timeout) {
+ this.timeout = timeout;
+ return this;
+ }
+
+ @Override
+ public long getTimeout() {
+ if (timeout == 0) {
+ timeout = TimeUnit.MINUTES.toMillis(2);
+ }
+ return timeout;
+ }
+
+ @Override
+ public CriticalAnalyzer addAction(CriticalAction action) {
+ this.actions.add(action);
+ return this;
+ }
+
+ @Override
+ public void check() {
+ boolean retry = true;
+ while (retry) {
+ try {
+ for (CriticalComponent component : components) {
+
+ if (component.isExpired(timeout)) {
+ fireAction(component);
+ // no need to keep running if there's already a component failed
+ return;
+ }
+ }
+ retry = false; // got to the end of the list, no need to retry
+ } catch (ConcurrentModificationException dontCare) {
+ // lets retry on the loop
+ }
+ }
+ }
+
+ private void fireAction(CriticalComponent component) {
+ for (CriticalAction action: actions) {
+ try {
+ action.run(component);
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ }
+
+ @Override
+ public void start() {
+
+ if (!running.tryAcquire()) {
+ // already running
+ return;
+ }
+
+ // we are not using any Thread Pool or any Scheduled Executors from the ArtemisServer
+ // as that would defeat the purpose,
+ // as in any deadlocks the schedulers may be starving for something not responding fast enough
+ thread = new Thread("Artemis Critical Analyzer") {
+ @Override
+ public void run() {
+ try {
+ while (true) {
+ if (running.tryAcquire(getCheckTime(), TimeUnit.MILLISECONDS)) {
+ running.release();
+ // this means that the server has been stopped as we could acquire the semaphore... returning now
+ break;
+ }
+ check();
+ }
+ } catch (InterruptedException interrupted) {
+ // i will just leave on that case
+ }
+ }
+ };
+
+ thread.setDaemon(true);
+
+ thread.start();
+ }
+
+ @Override
+ public void stop() {
+ if (!isStarted()) {
+ // already stopped, leaving
+ return;
+ }
+
+ running.release();
+
+ try {
+ if (thread != null) {
+ thread.join();
+ }
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public boolean isStarted() {
+ return running.availablePermits() == 0;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponent.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponent.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponent.java
new file mode 100644
index 0000000..a2459dd
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponent.java
@@ -0,0 +1,54 @@
+/**
+ * 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.utils.critical;
+
+
+/**
+ * A Critical component enters and leaves a critical state.
+ * You update a long every time you enter a critical path
+ * you update a different long with a System.currentMillis every time you leave that path.
+ *
+ * If the enterCritical > leaveCritical at any point, then you need to measure the timeout.
+ * if the system stops responding, then you have something irresponsive at the system.
+ */
+public interface CriticalComponent {
+
+ /**
+ * please save the time you entered here.
+ * Use volatile variables.
+ * No locks through anywhere.
+ */
+ default void enterCritical(int path) {
+ // I'm providing a default as some components may chose to calculate it themselves
+ }
+
+ /**
+ * please save the time you entered here
+ * Use volatile variables.
+ * No locks through anywhere.
+ */
+ default void leaveCritical(int path) {
+ // I'm providing a default as some components may chose to calculate it themselves
+ }
+
+ /**
+ * Is this Component expired at a given timeout.. on any of its paths.
+ * @param timeout
+ * @return -1 if it's ok, or the number of the path it failed
+ */
+ boolean isExpired(long timeout);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java
new file mode 100644
index 0000000..c1c5602
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java
@@ -0,0 +1,68 @@
+/**
+ * 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.utils.critical;
+
+/**
+ * This is not abstract as it could be used through aggregations or extensions.
+ * This is only good for cases where you call leave within the same thread as you called enter.
+ */
+public class CriticalComponentImpl implements CriticalComponent {
+
+ private final CriticalMeasure[] measures;
+ private final CriticalAnalyzer analyzer;
+
+ public CriticalComponentImpl(CriticalAnalyzer analyzer, int numberOfPaths) {
+ if (analyzer == null) {
+ analyzer = EmptyCriticalAnalyzer.getInstance();
+ }
+ this.analyzer = analyzer;
+
+ if (analyzer.isMeasuring()) {
+ measures = new CriticalMeasure[numberOfPaths];
+ for (int i = 0; i < numberOfPaths; i++) {
+ measures[i] = new CriticalMeasure();
+ }
+ } else {
+ measures = null;
+ }
+ }
+
+ @Override
+ public void enterCritical(int path) {
+ if (analyzer.isMeasuring()) {
+ measures[path].enterCritical();
+ }
+
+ }
+
+ @Override
+ public void leaveCritical(int path) {
+ if (analyzer.isMeasuring()) {
+ measures[path].leaveCritical();
+ }
+ }
+
+ @Override
+ public boolean isExpired(long timeout) {
+ for (int i = 0; i < measures.length; i++) {
+ if (measures[i].isExpired(timeout)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java
new file mode 100644
index 0000000..b853dc5
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java
@@ -0,0 +1,52 @@
+/**
+ * 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.utils.critical;
+
+import org.jboss.logging.Logger;
+
+public class CriticalMeasure {
+
+ private static final Logger logger = Logger.getLogger(CriticalMeasure.class);
+
+ private volatile long timeEnter;
+ private volatile long timeLeft;
+
+ public void enterCritical() {
+ timeEnter = System.currentTimeMillis();
+ }
+
+ public void leaveCritical() {
+ timeLeft = System.currentTimeMillis();
+ }
+
+ public boolean isExpired(long timeout) {
+ if (timeEnter > timeLeft) {
+ return System.currentTimeMillis() - timeEnter > timeout;
+ }
+
+ return false;
+ }
+
+ public long enterTime() {
+ return timeEnter;
+ }
+
+ public long leaveTime() {
+ return timeLeft;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/EmptyCriticalAnalyzer.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/EmptyCriticalAnalyzer.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/EmptyCriticalAnalyzer.java
new file mode 100644
index 0000000..4cf23a9
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/EmptyCriticalAnalyzer.java
@@ -0,0 +1,90 @@
+/**
+ * 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.utils.critical;
+
+public class EmptyCriticalAnalyzer implements CriticalAnalyzer {
+
+ private static final EmptyCriticalAnalyzer instance = new EmptyCriticalAnalyzer();
+
+ public static EmptyCriticalAnalyzer getInstance() {
+ return instance;
+ }
+
+ private EmptyCriticalAnalyzer() {
+ }
+
+ @Override
+ public void add(CriticalComponent component) {
+
+ }
+
+ @Override
+ public void remove(CriticalComponent component) {
+
+ }
+
+ @Override
+ public boolean isMeasuring() {
+ return false;
+ }
+
+ @Override
+ public void start() throws Exception {
+
+ }
+
+ @Override
+ public void stop() throws Exception {
+
+ }
+
+ @Override
+ public boolean isStarted() {
+ return false;
+ }
+
+ @Override
+ public CriticalAnalyzer setCheckTime(long timeout) {
+ return this;
+ }
+
+ @Override
+ public long getCheckTime() {
+ return 0;
+ }
+
+ @Override
+ public CriticalAnalyzer setTimeout(long timeout) {
+ return this;
+ }
+
+ @Override
+ public long getTimeout() {
+ return 0;
+ }
+
+ @Override
+ public CriticalAnalyzer addAction(CriticalAction action) {
+ return this;
+ }
+
+ @Override
+ public void check() {
+
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerTest.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerTest.java
new file mode 100644
index 0000000..638eb61
--- /dev/null
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerTest.java
@@ -0,0 +1,114 @@
+/**
+ * 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.utils.critical;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.utils.ThreadLeakCheckRule;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class CriticalAnalyzerTest {
+
+ @Rule
+ public ThreadLeakCheckRule rule = new ThreadLeakCheckRule();
+
+ private CriticalAnalyzer analyzer;
+
+ @After
+ public void tearDown() throws Exception {
+ if (analyzer != null) {
+ analyzer.stop();
+ }
+ }
+
+ @Test
+ public void testAction() throws Exception {
+ analyzer = new CriticalAnalyzerImpl().setTimeout(100).setCheckTime(50);
+ analyzer.add(new CriticalComponent() {
+ @Override
+ public boolean isExpired(long timeout) {
+ return true;
+ }
+ });
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ analyzer.start();
+
+ analyzer.addAction((CriticalComponent comp) -> {
+ System.out.println("component " + comp + " received");
+ latch.countDown();
+ });
+
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ analyzer.stop();
+ }
+
+ @Test
+ public void testActionOnImpl() throws Exception {
+ analyzer = new CriticalAnalyzerImpl().setTimeout(10).setCheckTime(5);
+ CriticalComponent component = new CriticalComponentImpl(analyzer, 2);
+ analyzer.add(component);
+
+ component.enterCritical(0);
+ component.leaveCritical(0);
+ component.enterCritical(1);
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ analyzer.start();
+
+ analyzer.addAction((CriticalComponent comp) -> {
+ System.out.println("component " + comp + " received");
+ latch.countDown();
+ });
+
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ analyzer.stop();
+ }
+
+ @Test
+ public void testNegative() throws Exception {
+ analyzer = new CriticalAnalyzerImpl().setTimeout(10).setCheckTime(5);
+ CriticalComponent component = new CriticalComponentImpl(analyzer, 1);
+ analyzer.add(component);
+
+ component.enterCritical(0);
+ component.leaveCritical(0);
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ analyzer.start();
+
+ analyzer.addAction((CriticalComponent comp) -> {
+ System.out.println("component " + comp + " received");
+ latch.countDown();
+ });
+
+ Assert.assertFalse(latch.await(100, TimeUnit.MILLISECONDS));
+
+ analyzer.stop();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
index 899dc2c..7e6684b 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
@@ -477,6 +477,12 @@ public final class ActiveMQDefaultConfiguration {
public static int DEFAULT_QUORUM_SIZE = -1;
+ public static final boolean DEFAULT_ANALYZE_CRITICAL = true;
+
+ public static final long DEFAULT_ANALYZE_CRITICAL_TIMEOUT = 120000;
+
+ public static final boolean DEFAULT_ANALYZE_CRITICAL_HALT = false;
+
/**
* If true then the ActiveMQ Artemis Server will make use of any Protocol Managers that are in available on the classpath. If false then only the core protocol will be available, unless in Embedded mode where users can inject their own Protocol Managers.
*/
@@ -1282,4 +1288,24 @@ public final class ActiveMQDefaultConfiguration {
public static int getDefaultQuorumSize() {
return DEFAULT_QUORUM_SIZE;
}
+
+
+ public static boolean getCriticalAnalyzer() {
+ return DEFAULT_ANALYZE_CRITICAL;
+ }
+
+ public static long getCriticalAnalyzerTimeout() {
+ return DEFAULT_ANALYZE_CRITICAL_TIMEOUT;
+ }
+
+ public static long getCriticalAnalyzerCheckPeriod(long timeout) {
+ // this will be 0, the implementation should return 1/2 of the configured critical timeout
+ return timeout / 2;
+ }
+
+ public static boolean getCriticalAnalyzerHalt() {
+ return DEFAULT_ANALYZE_CRITICAL_HALT;
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
index 6997443..e4c6c01 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
@@ -75,6 +75,23 @@ public interface Configuration {
Configuration parseSystemProperties(Properties properties) throws Exception;
+ boolean isCriticalAnalyzer();
+
+ Configuration setCriticalAnalyzer(boolean CriticalAnalyzer);
+
+ long getCriticalAnalyzerTimeout();
+
+ Configuration setCriticalAnalyzerTimeout(long timeout);
+
+ long getCriticalAnalyzerCheckPeriod();
+
+ Configuration setCriticalAnalyzerCheckPeriod(long checkPeriod);
+
+ boolean isCriticalAnalyzerHalt();
+
+ Configuration setCriticalAnalyzerHalt(boolean halt);
+
+
/**
* Returns whether this server is clustered. <br>
* {@code true} if {@link #getClusterConfigurations()} is not empty.
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
index 9195cff..e675606 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
@@ -294,6 +294,14 @@ public class ConfigurationImpl implements Configuration, Serializable {
private String internalNamingPrefix = ActiveMQDefaultConfiguration.getInternalNamingPrefix();
+ private boolean criticalAnalyzer = ActiveMQDefaultConfiguration.getCriticalAnalyzer();
+
+ private boolean criticalAnalyzerHalt = ActiveMQDefaultConfiguration.getCriticalAnalyzerHalt();
+
+ private long criticalAnalyzerTimeout = ActiveMQDefaultConfiguration.getCriticalAnalyzerTimeout();
+
+ private long criticalAnalyzerCheckPeriod = 0; // non set
+
/**
* Parent folder for all data folders.
*/
@@ -2064,6 +2072,53 @@ public class ConfigurationImpl implements Configuration, Serializable {
return this;
}
+ @Override
+ public boolean isCriticalAnalyzer() {
+ return criticalAnalyzer;
+ }
+
+ @Override
+ public Configuration setCriticalAnalyzer(boolean CriticalAnalyzer) {
+ this.criticalAnalyzer = CriticalAnalyzer;
+ return this;
+ }
+
+ @Override
+ public long getCriticalAnalyzerTimeout() {
+ return criticalAnalyzerTimeout;
+ }
+
+ @Override
+ public Configuration setCriticalAnalyzerTimeout(long timeout) {
+ this.criticalAnalyzerTimeout = timeout;
+ return this;
+ }
+
+ @Override
+ public long getCriticalAnalyzerCheckPeriod() {
+ if (criticalAnalyzerCheckPeriod <= 0) {
+ this.criticalAnalyzerCheckPeriod = ActiveMQDefaultConfiguration.getCriticalAnalyzerCheckPeriod(criticalAnalyzerTimeout);
+ }
+ return criticalAnalyzerCheckPeriod;
+ }
+
+ @Override
+ public Configuration setCriticalAnalyzerCheckPeriod(long checkPeriod) {
+ this.criticalAnalyzerCheckPeriod = checkPeriod;
+ return this;
+ }
+
+ @Override
+ public boolean isCriticalAnalyzerHalt() {
+ return criticalAnalyzerHalt;
+ }
+
+ @Override
+ public Configuration setCriticalAnalyzerHalt(boolean halt) {
+ this.criticalAnalyzerHalt = halt;
+ return this;
+ }
+
public static boolean checkoutDupCacheSize(final int windowSize, final int idCacheSize) {
final int msgNumInFlight = windowSize / DEFAULT_JMS_MESSAGE_SIZE;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
index a4a3487..9e1e807 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
@@ -608,6 +608,14 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
config.setNetworkCheckPingCommand(getString(e, "network-check-ping-command", config.getNetworkCheckPingCommand(), Validators.NO_CHECK));
+ config.setCriticalAnalyzer(getBoolean(e, "critical-analyzer", config.isCriticalAnalyzer()));
+
+ config.setCriticalAnalyzerTimeout(getLong(e, "critical-analyzer-timeout", config.getCriticalAnalyzerTimeout(), Validators.GE_ZERO));
+
+ config.setCriticalAnalyzerCheckPeriod(getLong(e, "critical-analyzer-check-period", config.getCriticalAnalyzerCheckPeriod(), Validators.GE_ZERO));
+
+ config.setCriticalAnalyzerHalt(getBoolean(e, "critical-analyzer-halt", config.isCriticalAnalyzerHalt()));
+
parseAddressSettings(e, config);
parseResourceLimits(e, config);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index dc399c1..58c86a0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -105,6 +105,8 @@ import org.apache.activemq.artemis.spi.core.protocol.MessagePersister;
import org.apache.activemq.artemis.utils.Base64;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.IDGenerator;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
+import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
import org.jboss.logging.Logger;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
@@ -121,7 +123,10 @@ import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalR
* <p>
* Using this class also ensures that locks are acquired in the right order, avoiding dead-locks.
*/
-public abstract class AbstractJournalStorageManager implements StorageManager {
+public abstract class AbstractJournalStorageManager extends CriticalComponentImpl implements StorageManager {
+
+ private static final int CRITICAL_PATHS = 1;
+ private static final int CRITICAL_STORE = 0;
private static final Logger logger = Logger.getLogger(AbstractJournalStorageManager.class);
@@ -188,17 +193,21 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
protected final Set<Long> largeMessagesToDelete = new HashSet<>();
public AbstractJournalStorageManager(final Configuration config,
+ final CriticalAnalyzer analyzer,
final ExecutorFactory executorFactory,
final ScheduledExecutorService scheduledExecutorService,
final ExecutorFactory ioExecutors) {
- this(config, executorFactory, scheduledExecutorService, ioExecutors, null);
+ this(config, analyzer, executorFactory, scheduledExecutorService, ioExecutors, null);
}
public AbstractJournalStorageManager(Configuration config,
+ CriticalAnalyzer analyzer,
ExecutorFactory executorFactory,
ScheduledExecutorService scheduledExecutorService,
ExecutorFactory ioExecutors,
IOCriticalErrorListener criticalErrorListener) {
+ super(analyzer, CRITICAL_PATHS);
+
this.executorFactory = executorFactory;
this.ioCriticalErrorListener = criticalErrorListener;
@@ -378,12 +387,14 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
@Override
public void readLock() {
+ enterCritical(CRITICAL_STORE);
storageManagerLock.readLock().lock();
}
@Override
public void readUnLock() {
storageManagerLock.readLock().unlock();
+ leaveCritical(CRITICAL_STORE);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
index 9923a3e..c819bb6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
@@ -32,24 +32,27 @@ import org.apache.activemq.artemis.jdbc.store.journal.JDBCJournalImpl;
import org.apache.activemq.artemis.jdbc.store.sql.GenericSQLProvider;
import org.apache.activemq.artemis.jdbc.store.sql.SQLProvider;
import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
public class JDBCJournalStorageManager extends JournalStorageManager {
private Connection connection;
public JDBCJournalStorageManager(Configuration config,
+ CriticalAnalyzer analyzer,
ExecutorFactory executorFactory,
ExecutorFactory ioExecutorFactory,
ScheduledExecutorService scheduledExecutorService) {
- super(config, executorFactory, scheduledExecutorService, ioExecutorFactory);
+ super(config, analyzer, executorFactory, scheduledExecutorService, ioExecutorFactory);
}
public JDBCJournalStorageManager(final Configuration config,
+ final CriticalAnalyzer analyzer,
final ScheduledExecutorService scheduledExecutorService,
final ExecutorFactory executorFactory,
final ExecutorFactory ioExecutorFactory,
final IOCriticalErrorListener criticalErrorListener) {
- super(config, executorFactory, scheduledExecutorService, ioExecutorFactory, criticalErrorListener);
+ super(config, analyzer, executorFactory, scheduledExecutorService, ioExecutorFactory, criticalErrorListener);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
index 148c1f0..2341a66 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
@@ -64,6 +64,7 @@ import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.jboss.logging.Logger;
public class JournalStorageManager extends AbstractJournalStorageManager {
@@ -85,29 +86,32 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
private ReplicationManager replicator;
public JournalStorageManager(final Configuration config,
+ final CriticalAnalyzer analyzer,
final ExecutorFactory executorFactory,
final ScheduledExecutorService scheduledExecutorService,
final ExecutorFactory ioExecutors) {
- this(config, executorFactory, scheduledExecutorService, ioExecutors, null);
+ this(config, analyzer, executorFactory, scheduledExecutorService, ioExecutors, null);
}
- public JournalStorageManager(final Configuration config, final ExecutorFactory executorFactory, final ExecutorFactory ioExecutors) {
- this(config, executorFactory, null, ioExecutors, null);
+ public JournalStorageManager(final Configuration config, CriticalAnalyzer analyzer, final ExecutorFactory executorFactory, final ExecutorFactory ioExecutors) {
+ this(config, analyzer, executorFactory, null, ioExecutors, null);
}
public JournalStorageManager(final Configuration config,
+ final CriticalAnalyzer analyzer,
final ExecutorFactory executorFactory,
final ScheduledExecutorService scheduledExecutorService,
final ExecutorFactory ioExecutors,
final IOCriticalErrorListener criticalErrorListener) {
- super(config, executorFactory, scheduledExecutorService, ioExecutors, criticalErrorListener);
+ super(config, analyzer, executorFactory, scheduledExecutorService, ioExecutors, criticalErrorListener);
}
public JournalStorageManager(final Configuration config,
+ final CriticalAnalyzer analyzer,
final ExecutorFactory executorFactory,
final ExecutorFactory ioExecutors,
final IOCriticalErrorListener criticalErrorListener) {
- super(config, executorFactory, null, ioExecutors, criticalErrorListener);
+ super(config, analyzer, executorFactory, null, ioExecutors, criticalErrorListener);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
index a3b93ea..643d2be 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
@@ -62,6 +62,7 @@ import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
/**
* This interface defines the internal interface of the ActiveMQ Artemis Server exposed to other components
@@ -108,6 +109,8 @@ public interface ActiveMQServer extends ServiceComponent {
NodeManager getNodeManager();
+ CriticalAnalyzer getCriticalAnalyzer();
+
/**
* @return
*/
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index f930f02..006ad23 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -1631,6 +1631,14 @@ public interface ActiveMQServerLogger extends BasicLogger {
@Message(id = 224075, value = "Cannot find pageTX id = {0}", format = Message.Format.MESSAGE_FORMAT)
void journalCannotFindPageTX(Long id);
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 224079, value = "The process for the virtual machine will be killed, as component {0} is not responsive", format = Message.Format.MESSAGE_FORMAT)
+ void criticalSystemHalt(Object component);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 224080, value = "The server process will now be stopped, as component {0} is not responsive", format = Message.Format.MESSAGE_FORMAT)
+ void criticalSystemShutdown(Object component);
+
@LogMessage(level = Logger.Level.INFO)
@Message(id = 224076, value = "UnDeploying address {0}", format = Message.Format.MESSAGE_FORMAT)
void undeployAddress(SimpleString addressName);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index 541b4d8..9a34837 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -31,8 +31,9 @@ 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.CriticalComponent;
-public interface Queue extends Bindable {
+public interface Queue extends Bindable,CriticalComponent {
int MAX_CONSUMERS_UNLIMITED = -1;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueFactory.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueFactory.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueFactory.java
index c2aeba4..7b377f6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueFactory.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueFactory.java
@@ -51,4 +51,8 @@ public interface QueueFactory {
* @param postOffice
*/
void setPostOffice(PostOffice postOffice);
+
+ default void queueRemoved(Queue queue) {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index 12848bb..4856d8b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -172,6 +172,10 @@ import org.apache.activemq.artemis.utils.TimeUtils;
import org.apache.activemq.artemis.utils.VersionLoader;
import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzerImpl;
+import org.apache.activemq.artemis.utils.critical.CriticalComponent;
+import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
import org.jboss.logging.Logger;
/**
@@ -316,6 +320,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
private final ActiveMQServer parentServer;
+
+ private final CriticalAnalyzer analyzer;
+
//todo think about moving this to the activation
private final List<SimpleString> scaledDownNodeIDs = new ArrayList<>();
@@ -426,6 +433,12 @@ public class ActiveMQServerImpl implements ActiveMQServer {
this.parentServer = parentServer;
this.serviceRegistry = serviceRegistry == null ? new ServiceRegistryImpl() : serviceRegistry;
+
+ if (configuration.isCriticalAnalyzer()) {
+ this.analyzer = new CriticalAnalyzerImpl();
+ } else {
+ this.analyzer = EmptyCriticalAnalyzer.getInstance();
+ }
}
@Override
@@ -479,12 +492,79 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
}
+ @Override
+ public CriticalAnalyzer getCriticalAnalyzer() {
+ return this.analyzer;
+ }
+
private void internalStart() throws Exception {
if (state != SERVER_STATE.STOPPED) {
logger.debug("Server already started!");
return;
}
+ /** Calling this for cases where the server was stopped and now is being restarted... failback, etc...*/
+ this.analyzer.clear();
+
+ this.getCriticalAnalyzer().setCheckTime(configuration.getCriticalAnalyzerCheckPeriod()).setTimeout(configuration.getCriticalAnalyzerTimeout());
+
+ if (configuration.isCriticalAnalyzer()) {
+ this.getCriticalAnalyzer().start();
+ }
+
+ this.getCriticalAnalyzer().addAction((CriticalComponent c) -> {
+
+ if (configuration.isCriticalAnalyzerHalt()) {
+ ActiveMQServerLogger.LOGGER.criticalSystemHalt(c);
+ } else {
+ ActiveMQServerLogger.LOGGER.criticalSystemShutdown(c);
+ }
+
+ threadDump();
+
+ // on the case of a critical failure, -1 cannot simply means forever.
+ // in case graceful is -1, we will set it to 30 seconds
+ long timeout = configuration.getGracefulShutdownTimeout() < 0 ? 30000 : configuration.getGracefulShutdownTimeout();
+
+ Thread notificationSender = new Thread() {
+ @Override
+ public void run() {
+ try {
+ callBrokerPlugins(hasBrokerPlugins() ? plugin -> plugin.criticalFailure(c) : null);
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ };
+
+ // I'm using a different thread here as we need to manage timeouts
+ notificationSender.start();
+
+ try {
+ notificationSender.join(timeout);
+ } catch (InterruptedException ignored) {
+ }
+
+ if (configuration.isCriticalAnalyzerHalt()) {
+ Runtime.getRuntime().halt(70); // Linux systems will have /usr/include/sysexits.h showing 70 as internal software error
+ } else {
+ // you can't stop from the check thread,
+ // nor can use an executor
+ Thread stopThread = new Thread() {
+ @Override
+ public void run() {
+ try {
+ ActiveMQServerImpl.this.stop();
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ };
+ stopThread.start();
+
+ }
+ });
+
configuration.parseSystemProperties();
startDate = new Date();
@@ -1061,6 +1141,12 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
}
+ try {
+ this.getCriticalAnalyzer().stop();
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ }
+
if (identity != null) {
ActiveMQServerLogger.LOGGER.serverStopped("identity=" + identity + ",version=" + getVersion().getFullVersion(), tempNodeID, getUptime());
} else {
@@ -2015,10 +2101,14 @@ public class ActiveMQServerImpl implements ActiveMQServer {
private StorageManager createStorageManager() {
if (configuration.isPersistenceEnabled()) {
if (configuration.getStoreConfiguration() != null && configuration.getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE) {
- return new JDBCJournalStorageManager(configuration, getScheduledPool(), executorFactory, ioExecutorFactory, shutdownOnCriticalIO);
+ JDBCJournalStorageManager journal = new JDBCJournalStorageManager(configuration, getCriticalAnalyzer(), getScheduledPool(), executorFactory, ioExecutorFactory, shutdownOnCriticalIO);
+ this.getCriticalAnalyzer().add(journal);
+ return journal;
} else {
// Default to File Based Storage Manager, (Legacy default configuration).
- return new JournalStorageManager(configuration, executorFactory, scheduledPool, ioExecutorFactory, shutdownOnCriticalIO);
+ JournalStorageManager journal = new JournalStorageManager(configuration, getCriticalAnalyzer(), executorFactory, scheduledPool, ioExecutorFactory, shutdownOnCriticalIO);
+ this.getCriticalAnalyzer().add(journal);
+ return journal;
}
}
return new NullStorageManager();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
index 8370839..192f25c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
@@ -32,6 +32,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.api.core.RoutingType;
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.settings.HierarchicalRepository;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.Transaction;
@@ -65,8 +66,9 @@ public class LastValueQueue extends QueueImpl {
final StorageManager storageManager,
final HierarchicalRepository<AddressSettings> addressSettingsRepository,
final Executor executor,
- final ActiveMQServer server) {
- super(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server);
+ final ActiveMQServer server,
+ final QueueFactory factory) {
+ super(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f16af753/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
index 3d8ceb1..33e66d1 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
@@ -75,10 +75,12 @@ public class QueueFactoryImpl implements QueueFactory {
final AddressSettings addressSettings = addressSettingsRepository.getMatch(config.address().toString());
final Queue queue;
if (addressSettings.isLastValueQueue()) {
- queue = new LastValueQueue(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server);
+ queue = new LastValueQueue(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this);
} else {
- queue = new QueueImpl(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server);
+ queue = new QueueImpl(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this);
}
+
+ server.getCriticalAnalyzer().add(queue);
return queue;
}
@@ -101,11 +103,18 @@ public class QueueFactoryImpl implements QueueFactory {
Queue queue;
if (addressSettings.isLastValueQueue()) {
- queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server);
+ queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this);
} else {
- queue = new QueueImpl(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server);
+ queue = new QueueImpl(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this);
}
+ server.getCriticalAnalyzer().add(queue);
+
return queue;
}
+
+ @Override
+ public void queueRemoved(Queue queue) {
+ server.getCriticalAnalyzer().remove(queue);
+ }
}