You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jb...@apache.org on 2016/11/23 16:14:33 UTC
[3/3] activemq-artemis git commit: ARTEMIS-473/ARTEMIS-863 Detect
network failures
ARTEMIS-473/ARTEMIS-863 Detect network failures
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/402f25be
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/402f25be
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/402f25be
Branch: refs/heads/master
Commit: 402f25be7dc5eda6e4dd1e8170e242415ce94fa8
Parents: 0e1fa2b
Author: Clebert Suconic <cl...@apache.org>
Authored: Thu Nov 17 10:01:31 2016 -0500
Committer: jbertram <jb...@apache.com>
Committed: Wed Nov 23 10:14:14 2016 -0600
----------------------------------------------------------------------
.../activemq/artemis/cli/commands/Create.java | 18 +-
.../activemq/artemis/cli/commands/Run.java | 2 +-
.../artemis/cli/commands/etc/broker.xml | 5 +-
.../commands/etc/commented-ping-settings.txt | 27 ++
.../artemis/cli/commands/etc/ping-settings.txt | 26 ++
.../apache/activemq/cli/test/ArtemisTest.java | 25 ++
.../activemq/cli/test/StreamClassPathTest.java | 3 +
.../core/server/ActiveMQScheduledComponent.java | 65 +++-
.../artemis/core/server/NetworkHealthCheck.java | 371 +++++++++++++++++++
.../artemis/utils/ActiveMQThreadFactory.java | 18 +-
.../utils/ActiveMQScheduledComponentTest.java | 62 ++++
.../artemis/utils/NetworkHealthTest.java | 216 +++++++++++
.../config/ActiveMQDefaultConfiguration.java | 31 ++
.../artemis/core/config/Configuration.java | 36 ++
.../artemis/core/config/ConfigurationUtils.java | 15 +-
.../core/config/impl/ConfigurationImpl.java | 101 ++++-
.../deployers/impl/FileConfigurationParser.java | 16 +-
.../core/persistence/StorageManager.java | 7 +-
.../journal/AbstractJournalStorageManager.java | 4 +-
.../impl/journal/JDBCJournalStorageManager.java | 2 +-
.../impl/journal/JournalStorageManager.java | 18 +-
.../impl/nullpm/NullStorageManager.java | 2 +-
.../artemis/core/server/ActiveMQServer.java | 10 +-
.../core/server/ActiveMQServerLogger.java | 6 +
.../core/server/cluster/ha/ReplicaPolicy.java | 18 +-
.../server/cluster/ha/ReplicatedPolicy.java | 18 +-
.../qourum/SharedNothingBackupQuorum.java | 20 +-
.../core/server/impl/ActiveMQServerImpl.java | 75 +++-
.../impl/SharedNothingBackupActivation.java | 11 +-
.../resources/schema/artemis-configuration.xsd | 60 +++
.../core/config/impl/FileConfigurationTest.java | 7 +
.../transaction/impl/TransactionImplTest.java | 2 +-
.../resources/ConfigurationTest-full-config.xml | 7 +
docs/user-manual/en/SUMMARY.md | 1 +
docs/user-manual/en/configuration-index.md | 5 +
docs/user-manual/en/network-isolation.md | 106 ++++++
.../cluster/failover/FailoverTestBase.java | 9 +-
.../NetworkIsolationReplicationTest.java | 134 +++++++
.../cluster/util/SameProcessActiveMQServer.java | 7 +-
.../cluster/util/TestableServer.java | 2 +
40 files changed, 1502 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java
index 77a2e68..1ae82b3 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java
@@ -63,6 +63,11 @@ public class Create extends InputAbstract {
private static final Integer MQTT_PORT = 1883;
+ /* **********************************************************************************
+ * Note for developers: These are tested at StreamClassPathTest on the unit test.
+ * This is to make sure maven or something else is not hiding these resources.
+ * ********************************************************************************** */
+
public static final String BIN_ARTEMIS_CMD = "bin/artemis.cmd";
public static final String BIN_ARTEMIS_SERVICE_EXE = "bin/artemis-service.exe";
public static final String BIN_ARTEMIS_SERVICE_XML = "bin/artemis-service.xml";
@@ -90,6 +95,8 @@ public class Create extends InputAbstract {
public static final String ETC_HORNETQ_ACCEPTOR_TXT = "etc/hornetq-acceptor.txt";
public static final String ETC_MQTT_ACCEPTOR_TXT = "etc/mqtt-acceptor.txt";
public static final String ETC_STOMP_ACCEPTOR_TXT = "etc/stomp-acceptor.txt";
+ public static final String ETC_PING_TXT = "etc/ping-settings.txt";
+ public static final String ETC_COMMENTED_PING_TXT = "etc/commented-ping-settings.txt";
@Arguments(description = "The instance directory to hold the broker's configuration and data. Path must be writable.", required = true)
File directory;
@@ -97,6 +104,9 @@ public class Create extends InputAbstract {
@Option(name = "--host", description = "The host name of the broker (Default: 0.0.0.0 or input if clustered)")
String host;
+ @Option(name = "--ping", description = "A comma separated string to be passed on to the broker config as network-check-list. The broker will shutdown when all these addresses are unreachable.")
+ String ping;
+
@Option(name = "--default-port", description = "The port number to use for the main 'artemis' acceptor (Default: 61616)")
int defaultPort = DEFAULT_PORT;
@@ -512,7 +522,6 @@ public class Create extends InputAbstract {
}
public Object run(ActionContext context) throws Exception {
-
if (forceLibaio && forceNIO) {
throw new RuntimeException("You can't specify --nio and --aio in the same execution.");
}
@@ -535,6 +544,13 @@ public class Create extends InputAbstract {
filters.put("${persistence-enabled}", isDisablePersistence() ? "false" : "true");
+ if (ping != null && !ping.isEmpty()) {
+ filters.put("${ping}", ping);
+ filters.put("${ping-config.settings}", applyFilters(readTextFile(ETC_PING_TXT), filters));
+ } else {
+ filters.put("${ping-config.settings}", readTextFile(ETC_COMMENTED_PING_TXT));
+ }
+
if (replicated) {
clustered = true;
filters.put("${replicated.settings}", applyFilters(readTextFile(ETC_REPLICATED_SETTINGS_TXT), filters));
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
index ff07106..0b9c8fd 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
@@ -151,7 +151,7 @@ public class Run extends LockAbstract {
}
}, 500, 500);
- Runtime.getRuntime().addShutdownHook(new Thread() {
+ Runtime.getRuntime().addShutdownHook(new Thread("shutdown-hook") {
@Override
public void run() {
try {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 9bcf368..23f2109 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
@@ -50,8 +50,9 @@ under the License.
<journal-min-files>2</journal-min-files>
<journal-pool-files>-1</journal-pool-files>
-${journal-buffer.settings}
-${connector-config.settings}
+
+${ping-config.settings}${journal-buffer.settings}${connector-config.settings}
+
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/commented-ping-settings.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/commented-ping-settings.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/commented-ping-settings.txt
new file mode 100644
index 0000000..c271dc9
--- /dev/null
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/commented-ping-settings.txt
@@ -0,0 +1,27 @@
+ <!--
+ You can specify the NIC you want to use to verify if the network
+ <network-check-NIC>theNickName</network-check-NIC>
+ -->
+
+ <!--
+ Use this to use an HTTP server to validate the network
+ <network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
+
+ <!-- <network-check-period>10000</network-check-period> -->
+ <!-- <network-check-timeout>1000</network-check-timeout> -->
+
+ <!-- this is a comma separated list, no spaces, just DNS or IPs
+ it should accept IPV6
+
+ Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
+ Using IPs that could eventually disappear or be partially visible may defeat the purpose.
+ You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
+ <!-- <network-check-list>10.0.0.1</network-check-list> -->
+
+ <!-- use this to customize the ping used for ipv4 addresses -->
+ <!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
+
+ <!-- use this to customize the ping used for ipv6 addresses -->
+ <!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
+
+
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/ping-settings.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/ping-settings.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/ping-settings.txt
new file mode 100644
index 0000000..90a2c00
--- /dev/null
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/ping-settings.txt
@@ -0,0 +1,26 @@
+ <!--
+ You can specify the NIC you want to use to verify if the network
+ <network-check-NIC>theNickName</network-check-NIC>
+ -->
+
+ <!--
+ Use this to use an HTTP server to validate the network
+ <network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
+
+ <network-check-period>10000</network-check-period>
+ <network-check-timeout>1000</network-check-timeout>
+
+ <!-- this is a comma separated list, no spaces, just DNS or IPs
+ it should accept IPV6
+
+ Warning: Make sure you understand your network topology as this is meant to check if your network is up.
+ Using IPs that could eventually disappear or be partially visible may defeat the purpose.
+ You can use a list of multiple IPs, any successful ping will make the server OK to continue running -->
+ <network-check-list>${ping}</network-check-list>
+
+ <!-- use this to customize the ping used for ipv4 addresses -->
+ <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command>
+
+ <!-- use this to customize the ping used for ipv6 addresses -->
+ <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command>
+
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 3d89aa8..ba02bd3 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
@@ -47,6 +47,8 @@ import org.apache.activemq.artemis.cli.commands.user.RemoveUser;
import org.apache.activemq.artemis.cli.commands.user.ResetUser;
import org.apache.activemq.artemis.cli.commands.util.SyncCalculation;
import org.apache.activemq.artemis.core.client.impl.ServerLocatorImpl;
+import org.apache.activemq.artemis.core.config.FileDeploymentManager;
+import org.apache.activemq.artemis.core.config.impl.FileConfiguration;
import org.apache.activemq.artemis.jlibaio.LibaioContext;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
@@ -605,6 +607,29 @@ public class ArtemisTest {
}
}
+
+ @Test
+ public void testPing() throws Exception {
+ File instanceFolder = temporaryFolder.newFolder("pingTest");
+
+ setupAuth(instanceFolder);
+ String queues = "q1,t2";
+ String topics = "t1,t2";
+
+ // This is usually set when run from the command line via artemis.profile
+ Run.setEmbedded(true);
+ Artemis.main("create", instanceFolder.getAbsolutePath(), "--force", "--silent", "--no-web", "--queues", queues, "--topics", topics, "--no-autotune", "--require-login", "--ping", "127.0.0.1");
+ System.setProperty("artemis.instance", instanceFolder.getAbsolutePath());
+
+ FileConfiguration fc = new FileConfiguration();
+ FileDeploymentManager deploymentManager = new FileDeploymentManager(new File(instanceFolder, "./etc/broker.xml").toURI().toString());
+ deploymentManager.addDeployable(fc);
+ deploymentManager.readConfiguration();
+
+ Assert.assertEquals("127.0.0.1", fc.getNetworkCheckList());
+
+ }
+
private void testCli(String... args) {
try {
Artemis.main(args);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-cli/src/test/java/org/apache/activemq/cli/test/StreamClassPathTest.java
----------------------------------------------------------------------
diff --git a/artemis-cli/src/test/java/org/apache/activemq/cli/test/StreamClassPathTest.java b/artemis-cli/src/test/java/org/apache/activemq/cli/test/StreamClassPathTest.java
index 133680b..f6d1fb3 100644
--- a/artemis-cli/src/test/java/org/apache/activemq/cli/test/StreamClassPathTest.java
+++ b/artemis-cli/src/test/java/org/apache/activemq/cli/test/StreamClassPathTest.java
@@ -54,6 +54,9 @@ public class StreamClassPathTest {
openStream(Create.ETC_MQTT_ACCEPTOR_TXT);
openStream(Create.ETC_HORNETQ_ACCEPTOR_TXT);
openStream(Create.ETC_STOMP_ACCEPTOR_TXT);
+ openStream(Create.ETC_PING_TXT);
+ openStream(Create.ETC_COMMENTED_PING_TXT);
+
}
private void openStream(String source) throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
index d62a744..7aeec1e 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
@@ -17,12 +17,16 @@
package org.apache.activemq.artemis.core.server;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.jboss.logging.Logger;
/**
@@ -31,7 +35,8 @@ import org.jboss.logging.Logger;
public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, Runnable {
private static final Logger logger = Logger.getLogger(ActiveMQScheduledComponent.class);
- private final ScheduledExecutorService scheduledExecutorService;
+ private ScheduledExecutorService scheduledExecutorService;
+ private boolean startedOwnScheduler;
private long period;
private long millisecondsPeriod;
private TimeUnit timeUnit;
@@ -50,24 +55,41 @@ public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, R
boolean onDemand) {
this.executor = executor;
this.scheduledExecutorService = scheduledExecutorService;
- if (this.scheduledExecutorService == null) {
- throw new NullPointerException("scheduled Executor is null");
- }
this.period = checkPeriod;
this.timeUnit = timeUnit;
this.onDemand = onDemand;
}
+ /**
+ * This is useful for cases where we want our own scheduler executor.
+ *
+ * @param checkPeriod
+ * @param timeUnit
+ * @param onDemand
+ */
+ public ActiveMQScheduledComponent(long checkPeriod, TimeUnit timeUnit, boolean onDemand) {
+ this(null, null, checkPeriod, timeUnit, onDemand);
+ }
+
@Override
public synchronized void start() {
if (future != null) {
+ // already started
return;
}
- this.millisecondsPeriod = timeUnit.convert(period, TimeUnit.MILLISECONDS);
+ if (scheduledExecutorService == null) {
+ scheduledExecutorService = new ScheduledThreadPoolExecutor(1, getThreadFactory());
+ startedOwnScheduler = true;
+
+ }
+
if (onDemand) {
return;
}
+
+ this.millisecondsPeriod = timeUnit.convert(period, TimeUnit.MILLISECONDS);
+
if (period >= 0) {
future = scheduledExecutorService.scheduleWithFixedDelay(runForScheduler, period, period, timeUnit);
} else {
@@ -75,6 +97,20 @@ public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, R
}
}
+ protected ActiveMQThreadFactory getThreadFactory() {
+ return new ActiveMQThreadFactory(this.getClass().getSimpleName() + "-scheduled-threads", false, getThisClassLoader());
+ }
+
+ private ClassLoader getThisClassLoader() {
+ return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
+ @Override
+ public ClassLoader run() {
+ return ActiveMQScheduledComponent.this.getClass().getClassLoader();
+ }
+ });
+
+ }
+
public void delay() {
int value = delayed.incrementAndGet();
if (value > 10) {
@@ -109,12 +145,15 @@ public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, R
@Override
public synchronized void stop() {
- if (future == null) {
- return; // no big deal
+ if (future != null) {
+ future.cancel(false);
+ future = null;
+ }
+ if (startedOwnScheduler) {
+ this.scheduledExecutorService.shutdownNow();
+ scheduledExecutorService = null;
+ startedOwnScheduler = false;
}
-
- future.cancel(false);
- future = null;
}
@@ -154,7 +193,11 @@ public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, R
final Runnable runForScheduler = new Runnable() {
@Override
public void run() {
- executor.execute(runForExecutor);
+ if (executor != null) {
+ executor.execute(runForExecutor);
+ } else {
+ runForExecutor.run();
+ }
}
};
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
new file mode 100644
index 0000000..8b7770d
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
@@ -0,0 +1,371 @@
+/**
+ * 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.core.server;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.URL;
+import java.net.URLConnection;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
+import org.apache.activemq.artemis.utils.ConcurrentHashSet;
+import org.jboss.logging.Logger;
+
+/**
+ * This will use {@link InetAddress#isReachable(int)} to determine if the network is alive.
+ * It will have a set of addresses, and if any address is reached the network will be considered alive.
+ */
+public class NetworkHealthCheck extends ActiveMQScheduledComponent {
+
+ private static final Logger logger = Logger.getLogger(NetworkHealthCheck.class);
+
+ private final Set<ActiveMQComponent> componentList = new ConcurrentHashSet<>();
+ private final Set<InetAddress> addresses = new ConcurrentHashSet<>();
+ private final Set<URL> urls = new ConcurrentHashSet<>();
+ private NetworkInterface networkInterface;
+
+ public static final String IPV6_DEFAULT_COMMAND = "ping6 -c 1 %2$s";
+
+ public static final String IPV4_DEFAULT_COMMAND = "ping -c 1 -t %d %s";
+
+ private String ipv4Command = IPV4_DEFAULT_COMMAND;
+
+ private String ipv6Command = IPV6_DEFAULT_COMMAND;
+
+ /**
+ * The timeout to be used on isReachable
+ */
+ private int networkTimeout;
+
+ public NetworkHealthCheck() {
+ this(null, 1000, 1000);
+ }
+
+ public NetworkHealthCheck(String nicName, long checkPeriod, int networkTimeout) {
+ super(null, null, checkPeriod, TimeUnit.MILLISECONDS, false);
+ this.networkTimeout = networkTimeout;
+ this.setNICName(nicName);
+ }
+
+ public NetworkHealthCheck setNICName(String nicName) {
+ NetworkInterface netToUse;
+ try {
+ if (nicName != null) {
+ netToUse = NetworkInterface.getByName(nicName);
+ } else {
+ netToUse = null;
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ netToUse = null;
+ }
+
+ this.networkInterface = netToUse;
+ return this;
+ }
+
+ public String getNICName() {
+ if (networkInterface != null) {
+ return networkInterface.getName();
+ } else {
+ return null;
+ }
+ }
+
+ public NetworkHealthCheck parseAddressList(String addressList) {
+ if (addressList != null) {
+ String[] addresses = addressList.split(",");
+
+ for (String address : addresses) {
+ try {
+ this.addAddress(InetAddress.getByName(address));
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ }
+
+ return this;
+ }
+
+ public NetworkHealthCheck parseURIList(String addressList) {
+ if (addressList != null) {
+ String[] addresses = addressList.split(",");
+
+ for (String address : addresses) {
+ try {
+ this.addURL(new URL(address));
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ }
+
+ return this;
+ }
+
+ @Override
+ protected ActiveMQThreadFactory getThreadFactory() {
+ return new ActiveMQThreadFactory("NetworkChecker", "Network-Checker-", false, getThisClassLoader());
+ }
+
+
+ private ClassLoader getThisClassLoader() {
+ return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
+ @Override
+ public ClassLoader run() {
+ return NetworkHealthCheck.this.getClass().getClassLoader();
+ }
+ });
+
+ }
+
+ public int getNetworkTimeout() {
+ return networkTimeout;
+ }
+
+ @Override
+ public NetworkHealthCheck setPeriod(long period) {
+ super.setPeriod(period);
+ return this;
+ }
+
+ @Override
+ public NetworkHealthCheck setTimeUnit(TimeUnit timeUnit) {
+ super.setTimeUnit(timeUnit);
+ return this;
+ }
+
+ public NetworkHealthCheck setNetworkTimeout(int networkTimeout) {
+ this.networkTimeout = networkTimeout;
+ return this;
+ }
+
+ public NetworkHealthCheck addComponent(ActiveMQComponent component) {
+ componentList.add(component);
+ checkStart();
+ return this;
+ }
+
+ public NetworkHealthCheck clearComponents() {
+ componentList.clear();
+ return this;
+ }
+
+ public NetworkHealthCheck addAddress(InetAddress address) {
+ if (!check(address)) {
+ logger.warn("Ping Address " + address + " wasn't reacheable");
+ }
+ addresses.add(address);
+
+ checkStart();
+ return this;
+ }
+
+ public NetworkHealthCheck removeAddress(InetAddress address) {
+ addresses.remove(address);
+ return this;
+ }
+
+ public NetworkHealthCheck clearAddresses() {
+ addresses.clear();
+ return this;
+ }
+
+ public NetworkHealthCheck addURL(URL url) {
+ if (!check(url)) {
+ logger.warn("Ping url " + url + " wasn't reacheable");
+ }
+ urls.add(url);
+ checkStart();
+ return this;
+ }
+
+ public NetworkHealthCheck removeURL(URL url) {
+ urls.remove(url);
+ return this;
+ }
+
+ public NetworkHealthCheck clearURL() {
+ urls.clear();
+ return this;
+ }
+
+ public String getIpv4Command() {
+ return ipv4Command;
+ }
+
+ public NetworkHealthCheck setIpv4Command(String ipv4Command) {
+ this.ipv4Command = ipv4Command;
+ return this;
+ }
+
+ public String getIpv6Command() {
+ return ipv6Command;
+ }
+
+ public NetworkHealthCheck setIpv6Command(String ipv6Command) {
+ this.ipv6Command = ipv6Command;
+ return this;
+ }
+
+ private void checkStart() {
+ if (!isStarted() && (!addresses.isEmpty() || !urls.isEmpty()) && !componentList.isEmpty()) {
+ start();
+ }
+ }
+
+ @Override
+ public void run() {
+ boolean healthy = check();
+
+ if (healthy) {
+ for (ActiveMQComponent component : componentList) {
+ if (!component.isStarted()) {
+ try {
+ logger.info("Network is healthy, starting service " + component);
+ component.start();
+ } catch (Exception e) {
+ logger.warn("Error starting component " + component, e);
+ }
+ }
+ }
+ } else {
+ for (ActiveMQComponent component : componentList) {
+ if (component.isStarted()) {
+ try {
+ logger.info("Network is unhealthy, stopping service " + component);
+ component.stop();
+ } catch (Exception e) {
+ logger.warn("Error stopping component " + component, e);
+ }
+ }
+ }
+ }
+
+ }
+
+ public boolean check() {
+ boolean isEmpty = true;
+ for (InetAddress address : addresses) {
+ isEmpty = false;
+ if (check(address)) {
+ return true;
+ }
+ }
+
+ for (URL url : urls) {
+ isEmpty = false;
+ if (check(url)) {
+ return true;
+ }
+ }
+
+ // This should return true if no checks were done, on this case it's empty
+ // This is tested by {@link NetworkHe
+ return isEmpty;
+ }
+
+ public boolean check(InetAddress address) {
+ try {
+ if (address.isReachable(networkInterface, 0, networkTimeout)) {
+ if (logger.isTraceEnabled()) {
+ logger.tracef(address + " OK");
+ }
+ return true;
+ } else {
+ return purePing(address);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ return false;
+ }
+ }
+
+ public boolean purePing(InetAddress address) throws IOException, InterruptedException {
+ long timeout = Math.max(1, TimeUnit.MILLISECONDS.toSeconds(networkTimeout));
+ // it did not work with a simple isReachable, it could be because there's no root access, so we will try ping executable
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("purePing on canonical address " + address.getCanonicalHostName());
+ }
+ ProcessBuilder processBuilder;
+ if (address instanceof Inet6Address) {
+ processBuilder = buildProcess(ipv6Command, timeout, address.getCanonicalHostName());
+ } else {
+ processBuilder = buildProcess(ipv4Command, timeout, address.getCanonicalHostName());
+ }
+
+ Process pingProcess = processBuilder.start();
+
+ readStream(pingProcess.getInputStream(), false);
+ readStream(pingProcess.getErrorStream(), true);
+
+ return pingProcess.waitFor() == 0;
+ }
+
+
+ private ProcessBuilder buildProcess(String expressionCommand, long timeout, String host) {
+ String command = String.format(expressionCommand, timeout, host);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("executing ping:: " + command);
+ }
+
+ ProcessBuilder builder = new ProcessBuilder(command.split(" "));
+
+ return builder;
+ }
+
+ private void readStream(InputStream stream, boolean error) throws IOException {
+ BufferedReader reader = new BufferedReader(new InputStreamReader(stream));
+
+ String inputLine;
+ while ((inputLine = reader.readLine()) != null) {
+ if (error) {
+ logger.warn(inputLine);
+ } else {
+ logger.trace(inputLine);
+ }
+ }
+
+ reader.close();
+ }
+
+ public boolean check(URL url) {
+ try {
+ URLConnection connection = url.openConnection();
+ connection.setReadTimeout(networkTimeout);
+ InputStream is = connection.getInputStream();
+ is.close();
+ return true;
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ return false;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
index 130a6d8..e5f76d3 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
@@ -36,6 +36,8 @@ public final class ActiveMQThreadFactory implements ThreadFactory {
private final AccessControlContext acc;
+ private final String prefix;
+
/**
* Construct a new instance. The access control context of the calling thread will be the one used to create
* new threads if a security manager is installed.
@@ -45,8 +47,22 @@ public final class ActiveMQThreadFactory implements ThreadFactory {
* @param tccl the context class loader of newly created threads
*/
public ActiveMQThreadFactory(final String groupName, final boolean daemon, final ClassLoader tccl) {
+ this(groupName, "Thread-", daemon, tccl);
+ }
+
+ /**
+ * Construct a new instance. The access control context of the calling thread will be the one used to create
+ * new threads if a security manager is installed.
+ *
+ * @param groupName the name of the thread group to assign threads to by default
+ * @param daemon whether the created threads should be daemon threads
+ * @param tccl the context class loader of newly created threads
+ */
+ public ActiveMQThreadFactory(final String groupName, String prefix, final boolean daemon, final ClassLoader tccl) {
group = new ThreadGroup(groupName + "-" + System.identityHashCode(this));
+ this.prefix = prefix;
+
this.threadPriority = Thread.NORM_PRIORITY;
this.tccl = tccl;
@@ -81,7 +97,7 @@ public final class ActiveMQThreadFactory implements ThreadFactory {
}
private Thread createThread(final Runnable command) {
- final Thread t = new Thread(group, command, "Thread-" + threadCount.getAndIncrement() + " (" + group.getName() + ")");
+ final Thread t = new Thread(group, command, prefix + threadCount.getAndIncrement() + " (" + group.getName() + ")");
t.setDaemon(daemon);
t.setPriority(threadPriority);
t.setContextClassLoader(tccl);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ActiveMQScheduledComponentTest.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ActiveMQScheduledComponentTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ActiveMQScheduledComponentTest.java
index 438f38e..0e64cae 100644
--- a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ActiveMQScheduledComponentTest.java
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ActiveMQScheduledComponentTest.java
@@ -17,6 +17,7 @@
package org.apache.activemq.artemis.utils;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -77,4 +78,65 @@ public class ActiveMQScheduledComponentTest {
Assert.assertTrue("just because one took a lot of time, it doesn't mean we can accumulate many, we got " + count + " executions", count.get() < 5);
}
+ @Test
+ public void testUsingOwnExecutors() throws Exception {
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ final ActiveMQScheduledComponent local = new ActiveMQScheduledComponent(10, TimeUnit.MILLISECONDS, false) {
+ @Override
+ public void run() {
+ latch.countDown();
+ }
+ };
+
+ local.start();
+ local.start(); // should be ok to call start again
+
+ try {
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ // re-scheduling the executor at a big interval..
+ // just to make sure it won't hung
+ local.setTimeUnit(TimeUnit.HOURS);
+ local.setPeriod(1);
+ } finally {
+ local.stop();
+ local.stop(); // should be ok to call stop again
+ }
+ }
+
+ @Test
+ public void testUsingOwnExecutorsOnDemand() throws Throwable {
+ final ReusableLatch latch = new ReusableLatch(1);
+
+ final ActiveMQScheduledComponent local = new ActiveMQScheduledComponent(10, TimeUnit.MILLISECONDS, true) {
+ @Override
+ public void run() {
+ latch.countDown();
+ }
+ };
+
+ local.start();
+ local.start(); // should be ok to call start again
+
+ try {
+
+ Assert.assertFalse(latch.await(20, TimeUnit.MILLISECONDS));
+
+ local.delay();
+ Assert.assertTrue(latch.await(20, TimeUnit.MILLISECONDS));
+ latch.setCount(1);
+
+ Assert.assertFalse(latch.await(20, TimeUnit.MILLISECONDS));
+
+ // re-scheduling the executor at a big interval..
+ // just to make sure it won't hung
+ local.setTimeUnit(TimeUnit.HOURS);
+ local.setPeriod(1);
+ } finally {
+ local.stop();
+ local.stop(); // calling stop again should not be an issue.
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NetworkHealthTest.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NetworkHealthTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NetworkHealthTest.java
new file mode 100644
index 0000000..7649912
--- /dev/null
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/NetworkHealthTest.java
@@ -0,0 +1,216 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class NetworkHealthTest {
+
+ private static final InetAddress INVALID_ADDRESS;
+
+ private static String IPV6_LOCAL = "::1";
+
+ static {
+ InetAddress address = null;
+ try {
+ address = InetAddress.getByName("203.0.113.1");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ INVALID_ADDRESS = address;
+ }
+
+ Set<NetworkHealthCheck> list = new HashSet<>();
+
+ NetworkHealthCheck addCheck(NetworkHealthCheck check) {
+ list.add(check);
+ return check;
+ }
+
+ HttpServer httpServer;
+
+ final ReusableLatch latch = new ReusableLatch(1);
+
+ ActiveMQComponent component = new ActiveMQComponent() {
+ boolean started = true;
+
+ @Override
+ public void start() throws Exception {
+ started = true;
+ latch.countDown();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ started = false;
+ latch.countDown();
+ }
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+ };
+
+ @Before
+ public void before() throws Exception {
+ latch.setCount(1);
+ }
+
+ private void startHTTPServer() throws IOException {
+ Assert.assertNull(httpServer);
+ InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8080);
+ httpServer = HttpServer.create(address, 100);
+ httpServer.start();
+ httpServer.createContext("/", new HttpHandler() {
+ @Override
+ public void handle(HttpExchange t) throws IOException {
+ String response = "<html><body><b>This is a unit test</b></body></html>";
+ t.sendResponseHeaders(200, response.length());
+ OutputStream os = t.getResponseBody();
+ os.write(response.getBytes());
+ os.close();
+ }
+ });
+ }
+
+ private void stopHTTPServer() {
+ if (httpServer != null) {
+ try {
+ httpServer.stop(0);
+ } catch (Throwable ignored) {
+ }
+ httpServer = null;
+ }
+ }
+
+ @After
+ public void after() {
+ stopHTTPServer();
+ for (NetworkHealthCheck check : this.list) {
+ check.stop();
+ }
+ }
+
+ @Test
+ public void testCheck6() throws Exception {
+ NetworkHealthCheck check = addCheck(new NetworkHealthCheck(null, 100, 100));
+ check.addComponent(component);
+
+ InetAddress address = InetAddress.getByName(IPV6_LOCAL);
+
+ Assert.assertTrue(address instanceof Inet6Address);
+
+ Assert.assertTrue(check.purePing(address));
+
+ Assert.assertTrue(check.check(address));
+
+ }
+
+ @Test
+ public void testPings() throws Exception {
+ doCheck("127.0.0.1");
+ }
+
+ private void doCheck(String localaddress) throws Exception {
+ NetworkHealthCheck check = addCheck(new NetworkHealthCheck(null, 100, 100));
+ check.addComponent(component);
+
+ // Any external IP, to make sure we would use a PING
+ InetAddress address = InetAddress.getByName(localaddress);
+
+ Assert.assertTrue(check.check(address));
+
+ Assert.assertTrue(check.purePing(address));
+
+ Assert.assertFalse(check.purePing(INVALID_ADDRESS));
+
+ }
+
+ @Test
+ public void testPingsIPV6() throws Exception {
+ doCheck(IPV6_LOCAL);
+ }
+
+ @Test
+ public void testCheckNoNodes() throws Exception {
+ NetworkHealthCheck check = addCheck(new NetworkHealthCheck());
+ Assert.assertTrue(check.check());
+ }
+
+ @Test
+ public void testCheckUsingHTTP() throws Exception {
+
+ startHTTPServer();
+
+ NetworkHealthCheck check = addCheck(new NetworkHealthCheck(null, 100, 1000));
+
+ Assert.assertTrue(check.check(new URL("http://localhost:8080")));
+
+ stopHTTPServer();
+
+ Assert.assertFalse(check.check(new URL("http://localhost:8080")));
+
+ check.addComponent(component);
+
+ URL url = new URL("http://localhost:8080");
+ Assert.assertFalse(check.check(url));
+
+ startHTTPServer();
+
+ Assert.assertTrue(check.check(url));
+
+ check.addURL(url);
+
+ Assert.assertFalse(latch.await(500, TimeUnit.MILLISECONDS));
+ Assert.assertTrue(component.isStarted());
+
+ // stopping the web server should stop the component
+ stopHTTPServer();
+
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+ Assert.assertFalse(component.isStarted());
+
+ latch.setCount(1);
+
+ startHTTPServer();
+
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+ Assert.assertTrue(component.isStarted());
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 33e7faa..7f503ba 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
@@ -440,6 +440,16 @@ public final class ActiveMQDefaultConfiguration {
public static final String DEFAULT_SYSTEM_PROPERTY_PREFIX = "brokerconfig.";
+ public static String DEFAULT_NETWORK_CHECK_LIST = null;
+
+ public static String DEFAULT_NETWORK_CHECK_URL_LIST = null;
+
+ public static long DEFAULT_NETWORK_CHECK_PERIOD = 5000;
+
+ public static int DEFAULT_NETWORK_CHECK_TIMEOUT = 1000;
+
+ public static String DEFAULT_NETWORK_CHECK_NIC = null;
+
/**
* 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.
*/
@@ -1181,4 +1191,25 @@ public final class ActiveMQDefaultConfiguration {
public static String getDefaultSystemPropertyPrefix() {
return DEFAULT_SYSTEM_PROPERTY_PREFIX;
}
+
+ public static String getDefaultNetworkCheckList() {
+ return DEFAULT_NETWORK_CHECK_LIST;
+ }
+
+ public static String getDefaultNetworkCheckURLList() {
+ return DEFAULT_NETWORK_CHECK_URL_LIST;
+ }
+
+ public static long getDefaultNetworkCheckPeriod() {
+ return DEFAULT_NETWORK_CHECK_PERIOD;
+ }
+
+ public static int getDefaultNetworkCheckTimeout() {
+ return DEFAULT_NETWORK_CHECK_TIMEOUT;
+ }
+
+ public static String getDefaultNetworkCheckNic() {
+ return DEFAULT_NETWORK_CHECK_NIC;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 8f4b97d..e2bedd1 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
@@ -1027,4 +1027,40 @@ public interface Configuration {
int getDiskScanPeriod();
+ /** A comma separated list of IPs we could use to validate if the network is UP.
+ * In case of none of these Ips are reached (if configured) the server will be shutdown. */
+ Configuration setNetworkCheckList(String list);
+
+ String getNetworkCheckList();
+
+ /** A comma separated list of URIs we could use to validate if the network is UP.
+ * In case of none of these Ips are reached (if configured) the server will be shutdown.
+ * The difference from networkCheckList is that we will use HTTP to make this validation. */
+ Configuration setNetworkCheckURLList(String uris);
+
+ String getNetworkCheckURLList();
+
+ /** The interval on which we will perform network checks. */
+ Configuration setNetworkCheckPeriod(long period);
+
+ long getNetworkCheckPeriod();
+
+ /** Time in ms for how long we should wait for a ping to finish. */
+ Configuration setNetworkCheckTimeout(int timeout);
+
+ int getNetworkCheckTimeout();
+
+ /** The NIC name to be used on network checks */
+ Configuration setNetworCheckNIC(String nic);
+
+ String getNetworkCheckNIC();
+
+ String getNetworkCheckPingCommand();
+
+ Configuration setNetworkCheckPingCommand(String command);
+
+ String getNetworkCheckPing6Command();
+
+ Configuration setNetworkCheckPing6Command(String command);
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ConfigurationUtils.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ConfigurationUtils.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ConfigurationUtils.java
index 4266860..beeb8da 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ConfigurationUtils.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ConfigurationUtils.java
@@ -24,6 +24,7 @@ import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreMasterPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreSlavePolicyConfiguration;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.ha.BackupPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ColocatedPolicy;
@@ -52,7 +53,7 @@ public final class ConfigurationUtils {
throw new ActiveMQIllegalStateException("Missing cluster-configuration for replication-clustername '" + replicationCluster + "'.");
}
- public static HAPolicy getHAPolicy(HAPolicyConfiguration conf) throws ActiveMQIllegalStateException {
+ public static HAPolicy getHAPolicy(HAPolicyConfiguration conf, ActiveMQServer server) throws ActiveMQIllegalStateException {
if (conf == null) {
return new LiveOnlyPolicy();
}
@@ -64,11 +65,11 @@ public final class ConfigurationUtils {
}
case REPLICATED: {
ReplicatedPolicyConfiguration pc = (ReplicatedPolicyConfiguration) conf;
- return new ReplicatedPolicy(pc.isCheckForLiveServer(), pc.getGroupName(), pc.getClusterName(), pc.getInitialReplicationSyncTimeout());
+ return new ReplicatedPolicy(pc.isCheckForLiveServer(), pc.getGroupName(), pc.getClusterName(), pc.getInitialReplicationSyncTimeout(), server.getNetworkHealthCheck());
}
case REPLICA: {
ReplicaPolicyConfiguration pc = (ReplicaPolicyConfiguration) conf;
- return new ReplicaPolicy(pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getGroupName(), pc.isRestartBackup(), pc.isAllowFailBack(), pc.getInitialReplicationSyncTimeout(), getScaleDownPolicy(pc.getScaleDownConfiguration()));
+ return new ReplicaPolicy(pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getGroupName(), pc.isRestartBackup(), pc.isAllowFailBack(), pc.getInitialReplicationSyncTimeout(), getScaleDownPolicy(pc.getScaleDownConfiguration()), server.getNetworkHealthCheck());
}
case SHARED_STORE_MASTER: {
SharedStoreMasterPolicyConfiguration pc = (SharedStoreMasterPolicyConfiguration) conf;
@@ -85,22 +86,22 @@ public final class ConfigurationUtils {
HAPolicy livePolicy;
//if null default to colocated
if (liveConf == null) {
- livePolicy = new ReplicatedPolicy();
+ livePolicy = new ReplicatedPolicy(server.getNetworkHealthCheck());
} else {
- livePolicy = getHAPolicy(liveConf);
+ livePolicy = getHAPolicy(liveConf, server);
}
HAPolicyConfiguration backupConf = pc.getBackupConfig();
BackupPolicy backupPolicy;
if (backupConf == null) {
if (livePolicy instanceof ReplicatedPolicy) {
- backupPolicy = new ReplicaPolicy();
+ backupPolicy = new ReplicaPolicy(server.getNetworkHealthCheck());
} else if (livePolicy instanceof SharedStoreMasterPolicy) {
backupPolicy = new SharedStoreSlavePolicy();
} else {
throw ActiveMQMessageBundle.BUNDLE.liveBackupMismatch();
}
} else {
- backupPolicy = (BackupPolicy) getHAPolicy(backupConf);
+ backupPolicy = (BackupPolicy) getHAPolicy(backupConf, server);
}
if ((livePolicy instanceof ReplicatedPolicy && !(backupPolicy instanceof ReplicaPolicy)) || (livePolicy instanceof SharedStoreMasterPolicy && !(backupPolicy instanceof SharedStoreSlavePolicy))) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 dce0998..e318874 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
@@ -57,6 +57,7 @@ import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
import org.apache.activemq.artemis.core.security.Role;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.JournalType;
+import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
import org.apache.activemq.artemis.core.server.SecuritySettingPlugin;
import org.apache.activemq.artemis.core.server.group.impl.GroupingHandlerConfiguration;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
@@ -261,6 +262,20 @@ public class ConfigurationImpl implements Configuration, Serializable {
private String systemPropertyPrefix = ActiveMQDefaultConfiguration.getDefaultSystemPropertyPrefix();
+ private String networkCheckList = ActiveMQDefaultConfiguration.getDefaultNetworkCheckList();
+
+ private String networkURLList = ActiveMQDefaultConfiguration.getDefaultNetworkCheckURLList();
+
+ private long networkCheckPeriod = ActiveMQDefaultConfiguration.getDefaultNetworkCheckPeriod();
+
+ private int networkCheckTimeout = ActiveMQDefaultConfiguration.getDefaultNetworkCheckTimeout();
+
+ private String networkCheckNIC = ActiveMQDefaultConfiguration.getDefaultNetworkCheckNic();
+
+ private String networkCheckPingCommand = NetworkHealthCheck.IPV4_DEFAULT_COMMAND;
+
+ private String networkCheckPing6Command = NetworkHealthCheck.IPV6_DEFAULT_COMMAND;
+
/**
* Parent folder for all data folders.
*/
@@ -279,7 +294,6 @@ public class ConfigurationImpl implements Configuration, Serializable {
return systemPropertyPrefix;
}
-
@Override
public Configuration parseSystemProperties() throws Exception {
parseSystemProperties(System.getProperties());
@@ -291,7 +305,6 @@ public class ConfigurationImpl implements Configuration, Serializable {
Map<String, Object> beanProperties = new HashMap<>();
-
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
if (entry.getKey().toString().startsWith(systemPropertyPrefix)) {
String key = entry.getKey().toString().substring(systemPropertyPrefix.length());
@@ -307,7 +320,6 @@ public class ConfigurationImpl implements Configuration, Serializable {
return this;
}
-
@Override
public boolean isClustered() {
return !getClusterConfigurations().isEmpty();
@@ -1874,6 +1886,89 @@ public class ConfigurationImpl implements Configuration, Serializable {
return this;
}
+ @Override
+ public ConfigurationImpl setNetworkCheckList(String list) {
+ this.networkCheckList = list;
+ return this;
+ }
+
+ @Override
+ public String getNetworkCheckList() {
+ return networkCheckList;
+ }
+
+ @Override
+ public ConfigurationImpl setNetworkCheckURLList(String urls) {
+ this.networkURLList = urls;
+ return this;
+ }
+
+ @Override
+ public String getNetworkCheckURLList() {
+ return networkURLList;
+ }
+
+ /**
+ * The interval on which we will perform network checks.
+ */
+ @Override
+ public ConfigurationImpl setNetworkCheckPeriod(long period) {
+ this.networkCheckPeriod = period;
+ return this;
+ }
+
+ @Override
+ public long getNetworkCheckPeriod() {
+ return this.networkCheckPeriod;
+ }
+
+ /**
+ * Time in ms for how long we should wait for a ping to finish.
+ */
+ @Override
+ public ConfigurationImpl setNetworkCheckTimeout(int timeout) {
+ this.networkCheckTimeout = timeout;
+ return this;
+ }
+
+ @Override
+ public int getNetworkCheckTimeout() {
+ return this.networkCheckTimeout;
+ }
+
+ @Override
+ public Configuration setNetworCheckNIC(String nic) {
+ this.networkCheckNIC = nic;
+ return this;
+ }
+
+ @Override
+ public String getNetworkCheckNIC() {
+ return networkCheckNIC;
+ }
+
+ @Override
+ public String getNetworkCheckPingCommand() {
+ return networkCheckPingCommand;
+ }
+
+ @Override
+ public ConfigurationImpl setNetworkCheckPingCommand(String command) {
+ this.networkCheckPingCommand = command;
+ return this;
+ }
+
+ @Override
+ public String getNetworkCheckPing6Command() {
+ return networkCheckPing6Command;
+ }
+
+ @Override
+ public Configuration setNetworkCheckPing6Command(String command) {
+ this.networkCheckPing6Command = command;
+ return this;
+ }
+
/**
* It will find the right location of a subFolder, related to artemisInstance
*/
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 3ea95d8..bd8aaf6 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
@@ -538,7 +538,21 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
config.setMemoryWarningThreshold(getInteger(e, "memory-warning-threshold", config.getMemoryWarningThreshold(), Validators.PERCENTAGE));
- config.setMemoryMeasureInterval(getLong(e, "memory-measure-interval", config.getMemoryMeasureInterval(), Validators.MINUS_ONE_OR_GT_ZERO)); // in
+ config.setMemoryMeasureInterval(getLong(e, "memory-measure-interval", config.getMemoryMeasureInterval(), Validators.MINUS_ONE_OR_GT_ZERO));
+
+ config.setNetworkCheckList(getString(e, "network-check-list", config.getNetworkCheckList(), Validators.NO_CHECK));
+
+ config.setNetworkCheckURLList(getString(e, "network-check-URL-list", config.getNetworkCheckURLList(), Validators.NO_CHECK));
+
+ config.setNetworkCheckPeriod(getLong(e, "network-check-period", config.getNetworkCheckPeriod(), Validators.GT_ZERO));
+
+ config.setNetworkCheckTimeout(getInteger(e, "network-check-timeout", config.getNetworkCheckTimeout(), Validators.GT_ZERO));
+
+ config.setNetworCheckNIC(getString(e, "network-check-NIC", config.getNetworkCheckNIC(), Validators.NO_CHECK));
+
+ config.setNetworkCheckPing6Command(getString(e, "network-check-ping6-command", config.getNetworkCheckPing6Command(), Validators.NO_CHECK));
+
+ config.setNetworkCheckPingCommand(getString(e, "network-check-ping-command", config.getNetworkCheckPingCommand(), Validators.NO_CHECK));
parseAddressSettings(e, config);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
index bbfec14..9675fe9 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
@@ -87,9 +87,12 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent {
void setContext(OperationContext context);
/**
- * @param ioCriticalError is the server being stopped due to an IO critical error
+ *
+ * @param ioCriticalError is the server being stopped due to an IO critical error.
+ * @param sendFailover this is to send the replication stopping in case of replication.
+ * @throws Exception
*/
- void stop(boolean ioCriticalError) throws Exception;
+ void stop(boolean ioCriticalError, boolean sendFailover) throws Exception;
// Message related operations
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 ecaa86e..191a89c 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
@@ -1437,7 +1437,7 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
@Override
public void stop() throws Exception {
- stop(false);
+ stop(false, true);
}
@Override
@@ -1454,7 +1454,7 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
protected abstract void performCachedLargeMessageDeletes();
@Override
- public synchronized void stop(boolean ioCriticalError) throws Exception {
+ public synchronized void stop(boolean ioCriticalError, boolean sendFailover) throws Exception {
if (!started) {
return;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 e4d401b..4e5c447 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
@@ -75,7 +75,7 @@ public class JDBCJournalStorageManager extends JournalStorageManager {
}
@Override
- public synchronized void stop(boolean ioCriticalError) throws Exception {
+ public synchronized void stop(boolean ioCriticalError, boolean sendFailover) throws Exception {
if (!started) {
return;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 c0ef93e..a1346ec 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
@@ -175,7 +175,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
@Override
public void stop() throws Exception {
- stop(false);
+ stop(false, true);
}
public boolean isReplicated() {
@@ -193,7 +193,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
}
@Override
- public synchronized void stop(boolean ioCriticalError) throws Exception {
+ public synchronized void stop(boolean ioCriticalError, boolean sendFailover) throws Exception {
if (!started) {
return;
}
@@ -224,12 +224,14 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
// and we want to ensure a stop here just in case
ReplicationManager replicatorInUse = replicator;
if (replicatorInUse != null) {
- final OperationContext token = replicator.sendLiveIsStopping(ReplicationLiveIsStoppingMessage.LiveStopping.FAIL_OVER);
- if (token != null) {
- try {
- token.waitCompletion(5000);
- } catch (Exception e) {
- // ignore it
+ if (sendFailover) {
+ final OperationContext token = replicator.sendLiveIsStopping(ReplicationLiveIsStoppingMessage.LiveStopping.FAIL_OVER);
+ if (token != null) {
+ try {
+ token.waitCompletion(5000);
+ } catch (Exception e) {
+ // ignore it
+ }
}
}
replicatorInUse.stop();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
index 3a2999e..b311888 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -507,7 +507,7 @@ public class NullStorageManager implements StorageManager {
}
@Override
- public void stop(final boolean ioCriticalError) throws Exception {
+ public void stop(final boolean ioCriticalError, boolean sendFailover) throws Exception {
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 aa1cb34..f8a5d75 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
@@ -32,6 +32,7 @@ import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.postoffice.PostOffice;
import org.apache.activemq.artemis.core.remoting.server.RemotingService;
+import org.apache.activemq.artemis.core.replication.ReplicationEndpoint;
import org.apache.activemq.artemis.core.replication.ReplicationManager;
import org.apache.activemq.artemis.core.security.Role;
import org.apache.activemq.artemis.core.security.SecurityAuth;
@@ -92,11 +93,18 @@ public interface ActiveMQServer extends ActiveMQComponent {
ActiveMQSecurityManager getSecurityManager();
+ NetworkHealthCheck getNetworkHealthCheck();
+
Version getVersion();
NodeManager getNodeManager();
/**
+ * @return
+ */
+ ReplicationEndpoint getReplicationEndpoint();
+
+ /**
* it will release hold a lock for the activation.
*/
void unlockActivation();
@@ -284,7 +292,7 @@ public interface ActiveMQServer extends ActiveMQComponent {
* @param filterString
* @param durable
* @throws org.apache.activemq.artemis.api.core.ActiveMQInvalidTransientQueueUseException if the shared queue already exists with a different {@code address} or {@code filterString}
- * @throws NullPointerException if {@code address} is {@code null}
+ * @throws NullPointerException if {@code address} is {@code null}
*/
void createSharedQueue(final SimpleString address,
final SimpleString name,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/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 2b58685..9b9570c 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
@@ -1265,6 +1265,12 @@ public interface ActiveMQServerLogger extends BasicLogger {
@Message(id = 222212, value = "Disk Full! Blocking message production on address ''{0}''. Clients will report blocked.", format = Message.Format.MESSAGE_FORMAT)
void blockingDiskFull(SimpleString addressName);
+ @LogMessage(level = Logger.Level.WARN)
+ @Message(id = 222213,
+ value = "There was an issue on the network, server is isolated!",
+ format = Message.Format.MESSAGE_FORMAT)
+ void serverIsolatedOnNetwork();
+
@LogMessage(level = Logger.Level.ERROR)
@Message(id = 224000, value = "Failure in initialisation", format = Message.Format.MESSAGE_FORMAT)
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicaPolicy.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicaPolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicaPolicy.java
index ba12677..89e7ab8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicaPolicy.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicaPolicy.java
@@ -19,6 +19,7 @@ package org.apache.activemq.artemis.core.server.cluster.ha;
import java.util.Map;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
import org.apache.activemq.artemis.core.server.impl.Activation;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.SharedNothingBackupActivation;
@@ -40,7 +41,10 @@ public class ReplicaPolicy extends BackupPolicy {
private ReplicatedPolicy replicatedPolicy;
- public ReplicaPolicy() {
+ private final NetworkHealthCheck networkHealthCheck;
+
+ public ReplicaPolicy(final NetworkHealthCheck networkHealthCheck) {
+ this.networkHealthCheck = networkHealthCheck;
}
public ReplicaPolicy(String clusterName,
@@ -49,7 +53,8 @@ public class ReplicaPolicy extends BackupPolicy {
boolean restartBackup,
boolean allowFailback,
long initialReplicationSyncTimeout,
- ScaleDownPolicy scaleDownPolicy) {
+ ScaleDownPolicy scaleDownPolicy,
+ NetworkHealthCheck networkHealthCheck) {
this.clusterName = clusterName;
this.maxSavedReplicatedJournalsSize = maxSavedReplicatedJournalsSize;
this.groupName = groupName;
@@ -57,16 +62,19 @@ public class ReplicaPolicy extends BackupPolicy {
this.allowFailback = allowFailback;
this.initialReplicationSyncTimeout = initialReplicationSyncTimeout;
this.scaleDownPolicy = scaleDownPolicy;
+ this.networkHealthCheck = networkHealthCheck;
}
public ReplicaPolicy(String clusterName,
int maxSavedReplicatedJournalsSize,
String groupName,
- ReplicatedPolicy replicatedPolicy) {
+ ReplicatedPolicy replicatedPolicy,
+ NetworkHealthCheck networkHealthCheck) {
this.clusterName = clusterName;
this.maxSavedReplicatedJournalsSize = maxSavedReplicatedJournalsSize;
this.groupName = groupName;
this.replicatedPolicy = replicatedPolicy;
+ this.networkHealthCheck = networkHealthCheck;
}
public String getClusterName() {
@@ -87,7 +95,7 @@ public class ReplicaPolicy extends BackupPolicy {
public ReplicatedPolicy getReplicatedPolicy() {
if (replicatedPolicy == null) {
- replicatedPolicy = new ReplicatedPolicy(false, allowFailback, initialReplicationSyncTimeout, groupName, clusterName, this);
+ replicatedPolicy = new ReplicatedPolicy(false, allowFailback, initialReplicationSyncTimeout, groupName, clusterName, this, networkHealthCheck);
}
return replicatedPolicy;
}
@@ -162,7 +170,7 @@ public class ReplicaPolicy extends BackupPolicy {
boolean wasLive,
Map<String, Object> activationParams,
ActiveMQServerImpl.ShutdownOnCriticalErrorListener shutdownOnCriticalIO) throws Exception {
- SharedNothingBackupActivation backupActivation = new SharedNothingBackupActivation(server, wasLive, activationParams, shutdownOnCriticalIO, this);
+ SharedNothingBackupActivation backupActivation = new SharedNothingBackupActivation(server, wasLive, activationParams, shutdownOnCriticalIO, this, networkHealthCheck);
backupActivation.init();
return backupActivation;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicatedPolicy.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicatedPolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicatedPolicy.java
index 9b64d81..f743219 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicatedPolicy.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/ReplicatedPolicy.java
@@ -19,6 +19,7 @@ package org.apache.activemq.artemis.core.server.cluster.ha;
import java.util.Map;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.LiveActivation;
import org.apache.activemq.artemis.core.server.impl.SharedNothingLiveActivation;
@@ -44,18 +45,23 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> {
* */
private ReplicaPolicy replicaPolicy;
- public ReplicatedPolicy() {
- replicaPolicy = new ReplicaPolicy(clusterName, -1, groupName, this);
+ private final NetworkHealthCheck networkHealthCheck;
+
+ public ReplicatedPolicy(NetworkHealthCheck networkHealthCheck) {
+ replicaPolicy = new ReplicaPolicy(clusterName, -1, groupName, this, networkHealthCheck);
+ this.networkHealthCheck = networkHealthCheck;
}
public ReplicatedPolicy(boolean checkForLiveServer,
String groupName,
String clusterName,
- long initialReplicationSyncTimeout) {
+ long initialReplicationSyncTimeout,
+ NetworkHealthCheck networkHealthCheck) {
this.checkForLiveServer = checkForLiveServer;
this.groupName = groupName;
this.clusterName = clusterName;
this.initialReplicationSyncTimeout = initialReplicationSyncTimeout;
+ this.networkHealthCheck = networkHealthCheck;
/*
* we create this with sensible defaults in case we start after a failover
* */
@@ -66,13 +72,15 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> {
long initialReplicationSyncTimeout,
String groupName,
String clusterName,
- ReplicaPolicy replicaPolicy) {
+ ReplicaPolicy replicaPolicy,
+ NetworkHealthCheck networkHealthCheck) {
this.checkForLiveServer = checkForLiveServer;
this.clusterName = clusterName;
this.groupName = groupName;
this.allowAutoFailBack = allowAutoFailBack;
this.initialReplicationSyncTimeout = initialReplicationSyncTimeout;
this.replicaPolicy = replicaPolicy;
+ this.networkHealthCheck = networkHealthCheck;
}
public boolean isCheckForLiveServer() {
@@ -114,7 +122,7 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> {
public ReplicaPolicy getReplicaPolicy() {
if (replicaPolicy == null) {
- replicaPolicy = new ReplicaPolicy(clusterName, -1, groupName, this);
+ replicaPolicy = new ReplicaPolicy(clusterName, -1, groupName, this, networkHealthCheck);
}
return replicaPolicy;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/402f25be/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/qourum/SharedNothingBackupQuorum.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/qourum/SharedNothingBackupQuorum.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/qourum/SharedNothingBackupQuorum.java
index 707849d..2f83c45 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/qourum/SharedNothingBackupQuorum.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/qourum/SharedNothingBackupQuorum.java
@@ -29,6 +29,7 @@ import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationLiveIsStoppingMessage;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
+import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
import org.apache.activemq.artemis.core.server.NodeManager;
public class SharedNothingBackupQuorum implements Quorum, SessionFailureListener {
@@ -52,6 +53,8 @@ public class SharedNothingBackupQuorum implements Quorum, SessionFailureListener
private CoreRemotingConnection connection;
+ private final NetworkHealthCheck networkHealthCheck;
+
/**
* This is a safety net in case the live sends the first {@link ReplicationLiveIsStoppingMessage}
* with code {@link org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationLiveIsStoppingMessage.LiveStopping#STOP_CALLED} and crashes before sending the second with
@@ -63,11 +66,13 @@ public class SharedNothingBackupQuorum implements Quorum, SessionFailureListener
public SharedNothingBackupQuorum(StorageManager storageManager,
NodeManager nodeManager,
- ScheduledExecutorService scheduledPool) {
+ ScheduledExecutorService scheduledPool,
+ NetworkHealthCheck networkHealthCheck) {
this.storageManager = storageManager;
this.scheduledPool = scheduledPool;
this.latch = new CountDownLatch(1);
this.nodeManager = nodeManager;
+ this.networkHealthCheck = networkHealthCheck;
}
private volatile BACKUP_ACTIVATION signal;
@@ -90,6 +95,9 @@ public class SharedNothingBackupQuorum implements Quorum, SessionFailureListener
//we may get called via multiple paths so need to guard
synchronized (decisionGuard) {
if (signal == BACKUP_ACTIVATION.FAIL_OVER) {
+ if (networkHealthCheck != null && !networkHealthCheck.check()) {
+ signal = BACKUP_ACTIVATION.FAILURE_REPLICATING;
+ }
return;
}
if (!isLiveDown()) {
@@ -102,8 +110,14 @@ public class SharedNothingBackupQuorum implements Quorum, SessionFailureListener
ActiveMQServerLogger.LOGGER.errorReConnecting(e);
}
}
- // live is assumed to be down, backup fails-over
- signal = BACKUP_ACTIVATION.FAIL_OVER;
+
+ if (networkHealthCheck != null && networkHealthCheck.check()) {
+ // live is assumed to be down, backup fails-over
+ signal = BACKUP_ACTIVATION.FAIL_OVER;
+ } else {
+ ActiveMQServerLogger.LOGGER.serverIsolatedOnNetwork();
+ signal = BACKUP_ACTIVATION.FAILURE_REPLICATING;
+ }
}
latch.countDown();
}