ARTEMIS-256 orchestrate failback deterministically The failback process needs to be deterministic rather than relying on various incarnations of Thread.sleep() at crucial points. Important aspects of this change include:
1) Make the initial replication synchronization process block at the very last step and wait for a response from the replica to ensure the replica has as the necessary data. This is a critical piece of knowledge during the failback process because it allows the soon-to-become-backup server to know for sure when it can shut itself down and allow the soon-to-become-live server to take over. Also, introduce a new configuration element called "initial-replication-sync-timeout" to conrol how long this blocking will occur. 2) Set the state of the server as 'LIVE' only after the server is fully started. This is necessary because once the soon-to-be-backup server shuts down it needs to know that the soon-to-be-live server has started fully before it restarts itself as the new backup. If the soon-to-be-backup server restarts before the soon-to-be-live is fully started then it won't actually become a backup server but instead will become a live server which will break the failback process. 3) Wait to receive the announcement of a backup server before failing-back. Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/ef5a9809 Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/ef5a9809 Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/ef5a9809 Branch: refs/heads/master Commit: ef5a9809f285d46e667dd81ae0040344d92c224b Parents: 30b3463 Author: jbertram <[email protected]> Authored: Wed Oct 14 12:07:17 2015 -0500 Committer: Clebert Suconic <[email protected]> Committed: Tue Oct 20 14:55:31 2015 -0400 ---------------------------------------------------------------------- .../config/ActiveMQDefaultConfiguration.java | 8 +-- .../core/protocol/core/impl/PacketImpl.java | 2 + .../artemis/core/config/ConfigurationUtils.java | 8 +-- .../config/ha/ReplicaPolicyConfiguration.java | 16 ++++- .../ha/ReplicatedPolicyConfiguration.java | 10 +++ .../SharedStoreMasterPolicyConfiguration.java | 7 +- .../ha/SharedStoreSlavePolicyConfiguration.java | 8 +-- .../deployers/impl/FileConfigurationParser.java | 8 +-- .../core/persistence/StorageManager.java | 5 +- .../impl/journal/JournalStorageManager.java | 5 +- .../impl/nullpm/NullStorageManager.java | 3 +- .../core/protocol/ServerPacketDecoder.java | 6 ++ .../wireformat/ReplicationResponseMessage.java | 6 +- .../ReplicationResponseMessageV2.java | 62 ++++++++++++++++++ .../wireformat/ReplicationStartSyncMessage.java | 12 ++++ .../core/replication/ReplicationEndpoint.java | 15 +++-- .../core/replication/ReplicationManager.java | 24 ++++++- .../core/server/ActiveMQMessageBundle.java | 3 + .../artemis/core/server/NodeManager.java | 4 +- .../core/server/cluster/ha/ReplicaPolicy.java | 21 ++++-- .../server/cluster/ha/ReplicatedPolicy.java | 24 +++++-- .../cluster/ha/SharedStoreMasterPolicy.java | 10 ++- .../cluster/ha/SharedStoreSlavePolicy.java | 13 ++-- .../core/server/impl/FileLockNodeManager.java | 34 +++++++++- .../core/server/impl/InVMNodeManager.java | 36 +++++++++-- .../impl/SharedNothingLiveActivation.java | 10 +-- .../impl/SharedStoreBackupActivation.java | 67 ++++++++++++-------- .../server/impl/SharedStoreLiveActivation.java | 3 +- .../resources/schema/artemis-configuration.xsd | 23 ++++++- .../config/impl/HAPolicyConfigurationTest.java | 33 +++++----- .../resources/colocated-hapolicy-config2.xml | 6 +- .../test/resources/replica-hapolicy-config.xml | 2 +- .../resources/replicated-hapolicy-config.xml | 1 + .../shared-store-master-hapolicy-config.xml | 2 - .../shared-store-slave-hapolicy-config.xml | 1 - .../shared-store-slave-hapolicy-config2.xml | 2 - .../shared-store-slave-hapolicy-config3.xml | 2 - docs/user-manual/en/ha.md | 54 +++++++--------- .../integration/cluster/NodeManagerAction.java | 2 +- .../cluster/failover/FailBackAutoTest.java | 4 +- .../cluster/failover/FailoverListenerTest.java | 4 +- .../cluster/failover/FailoverTestBase.java | 6 +- .../failover/LiveToLiveFailoverTest.java | 4 +- .../MultipleServerFailoverTestBase.java | 3 - .../failover/ReplicatedFailoverTest.java | 2 +- .../cluster/failover/SecurityFailoverTest.java | 2 +- .../cluster/failover/SharedStoreBackupTest.java | 4 +- .../cluster/util/BackupSyncDelay.java | 3 +- .../discovery/DiscoveryBaseTest.java | 24 ++++++- 49 files changed, 421 insertions(+), 193 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 21d97a3..6d8542a 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 @@ -374,8 +374,8 @@ public final class ActiveMQDefaultConfiguration { // Whether a server will automatically stop when another places a request to take over its place. The use case is when a regular server stops and its backup takes over its duties, later the main server restarts and requests the server (the former backup) to stop operating. private static boolean DEFAULT_ALLOW_AUTO_FAILBACK = true; - // if we have to start as a replicated server this is the delay to wait before fail-back occurs - private static long DEFAULT_FAILBACK_DELAY = 5000; + // When a replica comes online this is how long the replicating server will wait for a confirmation from the replica that the replication synchronization process is complete + private static long DEFAULT_INITIAL_REPLICATION_SYNC_TIMEOUT = 30000; // Will this backup server come live on a normal server shutdown private static boolean DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN = false; @@ -987,8 +987,8 @@ public final class ActiveMQDefaultConfiguration { /** * if we have to start as a replicated server this is the delay to wait before fail-back occurs */ - public static long getDefaultFailbackDelay() { - return DEFAULT_FAILBACK_DELAY; + public static long getDefaultInitialReplicationSyncTimeout() { + return DEFAULT_INITIAL_REPLICATION_SYNC_TIMEOUT; } /** http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java ---------------------------------------------------------------------- diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java index 06e259b..fffdec1 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java @@ -243,6 +243,8 @@ public class PacketImpl implements Packet { public static final byte SESS_BINDINGQUERY_RESP_V2 = -8; + public static final byte REPLICATION_RESPONSE_V2 = -9; + // Static -------------------------------------------------------- public PacketImpl(final byte type) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 e2a0a44..f1bb89c 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 @@ -64,19 +64,19 @@ public final class ConfigurationUtils { } case REPLICATED: { ReplicatedPolicyConfiguration pc = (ReplicatedPolicyConfiguration) conf; - return new ReplicatedPolicy(pc.isCheckForLiveServer(), pc.getGroupName(), pc.getClusterName()); + return new ReplicatedPolicy(pc.isCheckForLiveServer(), pc.getGroupName(), pc.getClusterName(), pc.getInitialReplicationSyncTimeout()); } case REPLICA: { ReplicaPolicyConfiguration pc = (ReplicaPolicyConfiguration) conf; - return new ReplicaPolicy(pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getGroupName(), pc.isRestartBackup(), pc.isAllowFailBack(), pc.getFailbackDelay(), getScaleDownPolicy(pc.getScaleDownConfiguration())); + return new ReplicaPolicy(pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getGroupName(), pc.isRestartBackup(), pc.isAllowFailBack(), pc.getInitialReplicationSyncTimeout(), getScaleDownPolicy(pc.getScaleDownConfiguration())); } case SHARED_STORE_MASTER: { SharedStoreMasterPolicyConfiguration pc = (SharedStoreMasterPolicyConfiguration) conf; - return new SharedStoreMasterPolicy(pc.getFailbackDelay(), pc.isFailoverOnServerShutdown()); + return new SharedStoreMasterPolicy(pc.isFailoverOnServerShutdown()); } case SHARED_STORE_SLAVE: { SharedStoreSlavePolicyConfiguration pc = (SharedStoreSlavePolicyConfiguration) conf; - return new SharedStoreSlavePolicy(pc.getFailbackDelay(), pc.isFailoverOnServerShutdown(), pc.isRestartBackup(), pc.isAllowFailBack(), getScaleDownPolicy(pc.getScaleDownConfiguration())); + return new SharedStoreSlavePolicy(pc.isFailoverOnServerShutdown(), pc.isRestartBackup(), pc.isAllowFailBack(), getScaleDownPolicy(pc.getScaleDownConfiguration())); } case COLOCATED: { ColocatedPolicyConfiguration pc = (ColocatedPolicyConfiguration) conf; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicaPolicyConfiguration.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicaPolicyConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicaPolicyConfiguration.java index 9663d24..17c83d4 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicaPolicyConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicaPolicyConfiguration.java @@ -37,7 +37,7 @@ public class ReplicaPolicyConfiguration implements HAPolicyConfiguration { * */ private boolean allowFailBack = false; - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); + private long initialReplicationSyncTimeout = ActiveMQDefaultConfiguration.getDefaultInitialReplicationSyncTimeout(); public ReplicaPolicyConfiguration() { } @@ -101,12 +101,22 @@ public class ReplicaPolicyConfiguration implements HAPolicyConfiguration { return this; } + @Deprecated public ReplicaPolicyConfiguration setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; return this; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; + } + + public long getInitialReplicationSyncTimeout() { + return initialReplicationSyncTimeout; + } + + public ReplicaPolicyConfiguration setInitialReplicationSyncTimeout(long initialReplicationSyncTimeout) { + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; + return this; } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicatedPolicyConfiguration.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicatedPolicyConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicatedPolicyConfiguration.java index ce62443..3b84bb7 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicatedPolicyConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/ReplicatedPolicyConfiguration.java @@ -27,6 +27,8 @@ public class ReplicatedPolicyConfiguration implements HAPolicyConfiguration { private String clusterName = null; + private long initialReplicationSyncTimeout = ActiveMQDefaultConfiguration.getDefaultInitialReplicationSyncTimeout(); + public ReplicatedPolicyConfiguration() { } @@ -61,4 +63,12 @@ public class ReplicatedPolicyConfiguration implements HAPolicyConfiguration { this.clusterName = clusterName; return this; } + + public long getInitialReplicationSyncTimeout() { + return initialReplicationSyncTimeout; + } + + public void setInitialReplicationSyncTimeout(long initialReplicationSyncTimeout) { + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; + } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreMasterPolicyConfiguration.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreMasterPolicyConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreMasterPolicyConfiguration.java index c868022..6668695 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreMasterPolicyConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreMasterPolicyConfiguration.java @@ -21,8 +21,6 @@ import org.apache.activemq.artemis.core.config.HAPolicyConfiguration; public class SharedStoreMasterPolicyConfiguration implements HAPolicyConfiguration { - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); - private boolean failoverOnServerShutdown = ActiveMQDefaultConfiguration.isDefaultFailoverOnServerShutdown(); public SharedStoreMasterPolicyConfiguration() { @@ -33,12 +31,13 @@ public class SharedStoreMasterPolicyConfiguration implements HAPolicyConfigurati return TYPE.SHARED_STORE_MASTER; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public SharedStoreMasterPolicyConfiguration setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; return this; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreSlavePolicyConfiguration.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreSlavePolicyConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreSlavePolicyConfiguration.java index 8e22077..f29c1d0 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreSlavePolicyConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/ha/SharedStoreSlavePolicyConfiguration.java @@ -22,8 +22,6 @@ import org.apache.activemq.artemis.core.config.ScaleDownConfiguration; public class SharedStoreSlavePolicyConfiguration implements HAPolicyConfiguration { - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); - private boolean failoverOnServerShutdown = ActiveMQDefaultConfiguration.isDefaultFailoverOnServerShutdown(); private boolean restartBackup = ActiveMQDefaultConfiguration.isDefaultRestartBackup(); @@ -76,13 +74,13 @@ public class SharedStoreSlavePolicyConfiguration implements HAPolicyConfiguratio return this; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public SharedStoreSlavePolicyConfiguration setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; return this; } - } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 0b081ad..0e75719 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 @@ -920,6 +920,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { configuration.setClusterName(getString(policyNode, "cluster-name", configuration.getClusterName(), Validators.NO_CHECK)); + configuration.setInitialReplicationSyncTimeout(getLong(policyNode, "initial-replication-sync-timeout", configuration.getInitialReplicationSyncTimeout(), Validators.GT_ZERO)); + return configuration; } @@ -932,7 +934,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { configuration.setAllowFailBack(getBoolean(policyNode, "allow-failback", configuration.isAllowFailBack())); - configuration.setFailbackDelay(getLong(policyNode, "failback-delay", configuration.getFailbackDelay(), Validators.GT_ZERO)); + configuration.setInitialReplicationSyncTimeout(getLong(policyNode, "initial-replication-sync-timeout", configuration.getInitialReplicationSyncTimeout(), Validators.GT_ZERO)); configuration.setClusterName(getString(policyNode, "cluster-name", configuration.getClusterName(), Validators.NO_CHECK)); @@ -948,8 +950,6 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { configuration.setFailoverOnServerShutdown(getBoolean(policyNode, "failover-on-shutdown", configuration.isFailoverOnServerShutdown())); - configuration.setFailbackDelay(getLong(policyNode, "failback-delay", configuration.getFailbackDelay(), Validators.GT_ZERO)); - return configuration; } @@ -960,8 +960,6 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { configuration.setFailoverOnServerShutdown(getBoolean(policyNode, "failover-on-shutdown", configuration.isFailoverOnServerShutdown())); - configuration.setFailbackDelay(getLong(policyNode, "failback-delay", configuration.getFailbackDelay(), Validators.GT_ZERO)); - configuration.setRestartBackup(getBoolean(policyNode, "restart-backup", configuration.isRestartBackup())); configuration.setScaleDownConfiguration(parseScaleDownConfig(policyNode)); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 76cb1bc..23dff8d 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 @@ -335,12 +335,13 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent { Journal getMessageJournal(); /** - * @see org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager#startReplication(org.apache.activemq.artemis.core.replication.ReplicationManager, org.apache.activemq.artemis.core.paging.PagingManager, String, boolean) + * @see org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager#startReplication(org.apache.activemq.artemis.core.replication.ReplicationManager, org.apache.activemq.artemis.core.paging.PagingManager, String, boolean, long) */ void startReplication(ReplicationManager replicationManager, PagingManager pagingManager, String nodeID, - boolean autoFailBack) throws Exception; + boolean autoFailBack, + long initialReplicationSyncTimeout) throws Exception; /** * Write message to page if we are paging. http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 440bd62..cac2c00 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 @@ -305,7 +305,8 @@ public class JournalStorageManager implements StorageManager { public void startReplication(ReplicationManager replicationManager, PagingManager pagingManager, String nodeID, - final boolean autoFailBack) throws Exception { + final boolean autoFailBack, + long initialReplicationSyncTimeout) throws Exception { if (!started) { throw new IllegalStateException("JournalStorageManager must be started..."); } @@ -376,7 +377,7 @@ public class JournalStorageManager implements StorageManager { storageManagerLock.writeLock().lock(); try { if (replicator != null) { - replicator.sendSynchronizationDone(nodeID); + replicator.sendSynchronizationDone(nodeID, initialReplicationSyncTimeout); performCachedLargeMessageDeletes(); } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 e6c1fe0..289cb77 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 @@ -473,7 +473,8 @@ public class NullStorageManager implements StorageManager { public void startReplication(final ReplicationManager replicationManager, final PagingManager pagingManager, final String nodeID, - final boolean autoFailBack) throws Exception { + final boolean autoFailBack, + long initialReplicationSyncTimeout) throws Exception { // no-op } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java index 796fb60..4457808 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java @@ -35,6 +35,7 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REP import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_PAGE_WRITE; import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_PREPARE; import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_RESPONSE; +import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_RESPONSE_V2; import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND; import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_LARGE; import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SCALEDOWN_ANNOUNCEMENT; @@ -64,6 +65,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Replicatio import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPageWriteMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPrepareMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessage; +import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessageV2; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationSyncFileMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ScaleDownAnnounceMessage; @@ -120,6 +122,10 @@ public class ServerPacketDecoder extends ClientPacketDecoder { packet = new ReplicationResponseMessage(); break; } + case REPLICATION_RESPONSE_V2: { + packet = new ReplicationResponseMessageV2(); + break; + } case REPLICATION_PAGE_WRITE: { packet = new ReplicationPageWriteMessage(); break; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessage.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessage.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessage.java index ac06997..c7eff85 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessage.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessage.java @@ -18,9 +18,13 @@ package org.apache.activemq.artemis.core.protocol.core.impl.wireformat; import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl; -public final class ReplicationResponseMessage extends PacketImpl { +public class ReplicationResponseMessage extends PacketImpl { public ReplicationResponseMessage() { super(PacketImpl.REPLICATION_RESPONSE); } + + public ReplicationResponseMessage(byte replicationResponseV2) { + super(replicationResponseV2); + } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessageV2.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessageV2.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessageV2.java new file mode 100644 index 0000000..146a3de --- /dev/null +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationResponseMessageV2.java @@ -0,0 +1,62 @@ +/* + * 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.protocol.core.impl.wireformat; + +import org.apache.activemq.artemis.api.core.ActiveMQBuffer; +import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl; + +public final class ReplicationResponseMessageV2 extends ReplicationResponseMessage { + boolean synchronizationIsFinishedAcknowledgement = false; + + public ReplicationResponseMessageV2(final boolean synchronizationIsFinishedAcknowledgement) { + super(REPLICATION_RESPONSE_V2); + + this.synchronizationIsFinishedAcknowledgement = synchronizationIsFinishedAcknowledgement; + } + + public ReplicationResponseMessageV2() { + super(PacketImpl.REPLICATION_RESPONSE_V2); + } + + public boolean isSynchronizationIsFinishedAcknowledgement() { + return synchronizationIsFinishedAcknowledgement; + } + + public void setSynchronizationIsFinishedAcknowledgement(boolean synchronizationIsFinishedAcknowledgement) { + this.synchronizationIsFinishedAcknowledgement = synchronizationIsFinishedAcknowledgement; + } + + @Override + public void encodeRest(final ActiveMQBuffer buffer) { + super.encodeRest(buffer); + buffer.writeBoolean(synchronizationIsFinishedAcknowledgement); + } + + @Override + public void decodeRest(final ActiveMQBuffer buffer) { + super.decodeRest(buffer); + synchronizationIsFinishedAcknowledgement = buffer.readBoolean(); + } + + @Override + public String toString() { + StringBuffer buf = new StringBuffer(getParentString()); + buf.append(", synchronizationIsFinishedAcknowledgement=" + synchronizationIsFinishedAcknowledgement); + buf.append("]"); + return buf.toString(); + } +} http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java index d6d6753..56c9461 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java @@ -203,4 +203,16 @@ public class ReplicationStartSyncMessage extends PacketImpl { return false; return true; } + + @Override + public String toString() { + StringBuffer buf = new StringBuffer(getParentString()); + buf.append(", synchronizationIsFinished=" + synchronizationIsFinished); + buf.append(", dataType=" + dataType); + buf.append(", nodeID=" + nodeID); + buf.append(", ids=" + Arrays.toString(ids)); + buf.append(", allowsAutoFailBack=" + allowsAutoFailBack); + buf.append("]"); + return buf.toString(); + } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java index c96a10f..c79e572 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java @@ -68,6 +68,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Replicatio import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPageWriteMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPrepareMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessage; +import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessageV2; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage.SyncDataType; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationSyncFileMessage; @@ -196,7 +197,7 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon handleLargeMessageEnd((ReplicationLargeMessageEndMessage) packet); } else if (type == PacketImpl.REPLICATION_START_FINISH_SYNC) { - handleStartReplicationSynchronization((ReplicationStartSyncMessage) packet); + response = handleStartReplicationSynchronization((ReplicationStartSyncMessage) packet); } else if (type == PacketImpl.REPLICATION_SYNC_FILE) { handleReplicationSynchronization((ReplicationSyncFileMessage) packet); @@ -476,19 +477,23 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon * * @param packet * @throws Exception + * @return if the incoming packet indicates the synchronization is finished then return an acknowledgement otherwise + * return an empty response */ - private void handleStartReplicationSynchronization(final ReplicationStartSyncMessage packet) throws Exception { + private ReplicationResponseMessageV2 handleStartReplicationSynchronization(final ReplicationStartSyncMessage packet) throws Exception { + ReplicationResponseMessageV2 replicationResponseMessage = new ReplicationResponseMessageV2(); if (activation.isRemoteBackupUpToDate()) { throw ActiveMQMessageBundle.BUNDLE.replicationBackupUpToDate(); } synchronized (this) { if (!started) - return; + return replicationResponseMessage; if (packet.isSynchronizationFinished()) { finishSynchronization(packet.getNodeID()); - return; + replicationResponseMessage.setSynchronizationIsFinishedAcknowledgement(true); + return replicationResponseMessage; } switch (packet.getDataType()) { @@ -523,6 +528,8 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon throw ActiveMQMessageBundle.BUNDLE.replicationUnhandledDataType(); } } + + return replicationResponseMessage; } private void handleLargeMessageEnd(final ReplicationLargeMessageEndMessage packet) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java index fa2b72c..d276474 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java @@ -58,12 +58,15 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Replicatio import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPageEventMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPageWriteMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationPrepareMessage; +import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessageV2; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage; import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationSyncFileMessage; import org.apache.activemq.artemis.core.server.ActiveMQComponent; +import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle; import org.apache.activemq.artemis.core.server.ActiveMQServerLogger; import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; import org.apache.activemq.artemis.utils.ExecutorFactory; +import org.apache.activemq.artemis.utils.ReusableLatch; /** * Manages replication tasks on the live server (that is the live server side of a "remote backup" @@ -116,6 +119,8 @@ public final class ReplicationManager implements ActiveMQComponent { private volatile boolean inSync = true; + private final ReusableLatch synchronizationIsFinishedAcknowledgement = new ReusableLatch(0); + /** * @param remotingConnection */ @@ -392,8 +397,14 @@ public final class ReplicationManager implements ActiveMQComponent { private final class ResponseHandler implements ChannelHandler { public void handlePacket(final Packet packet) { - if (packet.getType() == PacketImpl.REPLICATION_RESPONSE) { + if (packet.getType() == PacketImpl.REPLICATION_RESPONSE || packet.getType() == PacketImpl.REPLICATION_RESPONSE_V2) { replicated(); + if (packet.getType() == PacketImpl.REPLICATION_RESPONSE_V2) { + ReplicationResponseMessageV2 replicationResponseMessage = (ReplicationResponseMessageV2) packet; + if (replicationResponseMessage.isSynchronizationIsFinishedAcknowledgement()) { + synchronizationIsFinishedAcknowledgement.countDown(); + } + } } } @@ -534,9 +545,18 @@ public final class ReplicationManager implements ActiveMQComponent { * * @param nodeID */ - public void sendSynchronizationDone(String nodeID) { + public void sendSynchronizationDone(String nodeID, long initialReplicationSyncTimeout) { if (enabled) { + synchronizationIsFinishedAcknowledgement.countUp(); sendReplicatePacket(new ReplicationStartSyncMessage(nodeID)); + try { + if (!synchronizationIsFinishedAcknowledgement.await(initialReplicationSyncTimeout)) { + throw ActiveMQMessageBundle.BUNDLE.replicationSynchronizationTimeout(initialReplicationSyncTimeout); + } + } + catch (InterruptedException e) { + ActiveMQServerLogger.LOGGER.debug(e); + } inSync = false; } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java index e4402f4..3d57ade 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java @@ -359,4 +359,7 @@ public interface ActiveMQMessageBundle { @Message(id = 119113, value = "Invalid message load balancing type {0}", format = Message.Format.MESSAGE_FORMAT) IllegalArgumentException invalidMessageLoadBalancingType(String val); + + @Message(id = 119114, value = "Replication synchronization process timed out after waiting {0} milliseconds", format = Message.Format.MESSAGE_FORMAT) + IllegalStateException replicationSynchronizationTimeout(long timeout); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/NodeManager.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/NodeManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/NodeManager.java index 421daaa..9d54b92 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/NodeManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/NodeManager.java @@ -51,9 +51,11 @@ public abstract class NodeManager implements ActiveMQComponent { public abstract void awaitLiveNode() throws Exception; + public abstract void awaitLiveStatus() throws Exception; + public abstract void startBackup() throws Exception; - public abstract void startLiveNode() throws Exception; + public abstract ActivateCallback startLiveNode() throws Exception; public abstract void pauseLiveServer() throws Exception; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 6ec85b4..c32b446 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 @@ -36,7 +36,7 @@ public class ReplicaPolicy extends BackupPolicy { //used if we create a replicated policy for when we become live. private boolean allowFailback = ActiveMQDefaultConfiguration.isDefaultAllowAutoFailback(); - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); + private long initialReplicationSyncTimeout = ActiveMQDefaultConfiguration.getDefaultInitialReplicationSyncTimeout(); private ReplicatedPolicy replicatedPolicy; @@ -48,14 +48,14 @@ public class ReplicaPolicy extends BackupPolicy { String groupName, boolean restartBackup, boolean allowFailback, - long failbackDelay, + long initialReplicationSyncTimeout, ScaleDownPolicy scaleDownPolicy) { this.clusterName = clusterName; this.maxSavedReplicatedJournalsSize = maxSavedReplicatedJournalsSize; this.groupName = groupName; this.restartBackup = restartBackup; this.allowFailback = allowFailback; - this.failbackDelay = failbackDelay; + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; this.scaleDownPolicy = scaleDownPolicy; } @@ -87,7 +87,7 @@ public class ReplicaPolicy extends BackupPolicy { public ReplicatedPolicy getReplicatedPolicy() { if (replicatedPolicy == null) { - replicatedPolicy = new ReplicatedPolicy(false, allowFailback, failbackDelay, groupName, clusterName, this); + replicatedPolicy = new ReplicatedPolicy(false, allowFailback, initialReplicationSyncTimeout, groupName, clusterName, this); } return replicatedPolicy; } @@ -137,12 +137,21 @@ public class ReplicaPolicy extends BackupPolicy { this.allowFailback = allowFailback; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public void setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; + } + + public long getInitialReplicationSyncTimeout() { + return initialReplicationSyncTimeout; + } + + public void setInitialReplicationSyncTimeout(long initialReplicationSyncTimeout) { + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; } @Override http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/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 85fde90..295a862 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 @@ -31,14 +31,14 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> { private String clusterName; + private long initialReplicationSyncTimeout = ActiveMQDefaultConfiguration.getDefaultInitialReplicationSyncTimeout(); + /* * these are only set by the ReplicaPolicy after failover to decide if the live server can failback, these should not * be exposed in configuration. * */ private boolean allowAutoFailBack = ActiveMQDefaultConfiguration.isDefaultAllowAutoFailback(); - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); - /* * this are only used as the policy when the server is started as a live after a failover * */ @@ -48,10 +48,11 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> { replicaPolicy = new ReplicaPolicy(clusterName, -1, groupName, this); } - public ReplicatedPolicy(boolean checkForLiveServer, String groupName, String clusterName) { + public ReplicatedPolicy(boolean checkForLiveServer, String groupName, String clusterName, long initialReplicationSyncTimeout) { this.checkForLiveServer = checkForLiveServer; this.groupName = groupName; this.clusterName = clusterName; + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; /* * we create this with sensible defaults in case we start after a failover * */ @@ -59,7 +60,7 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> { public ReplicatedPolicy(boolean checkForLiveServer, boolean allowAutoFailBack, - long failbackDelay, + long initialReplicationSyncTimeout, String groupName, String clusterName, ReplicaPolicy replicaPolicy) { @@ -67,7 +68,7 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> { this.clusterName = clusterName; this.groupName = groupName; this.allowAutoFailBack = allowAutoFailBack; - this.failbackDelay = failbackDelay; + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; this.replicaPolicy = replicaPolicy; } @@ -83,12 +84,21 @@ public class ReplicatedPolicy implements HAPolicy<LiveActivation> { return allowAutoFailBack; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public void setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; + } + + public long getInitialReplicationSyncTimeout() { + return initialReplicationSyncTimeout; + } + + public void setInitialReplicationSyncTimeout(long initialReplicationSyncTimeout) { + this.initialReplicationSyncTimeout = initialReplicationSyncTimeout; } public String getClusterName() { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreMasterPolicy.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreMasterPolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreMasterPolicy.java index 653cd93..d1fcb65 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreMasterPolicy.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreMasterPolicy.java @@ -25,8 +25,6 @@ import java.util.Map; public class SharedStoreMasterPolicy implements HAPolicy<LiveActivation> { - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); - private boolean failoverOnServerShutdown = ActiveMQDefaultConfiguration.isDefaultFailoverOnServerShutdown(); private SharedStoreSlavePolicy sharedStoreSlavePolicy; @@ -34,17 +32,17 @@ public class SharedStoreMasterPolicy implements HAPolicy<LiveActivation> { public SharedStoreMasterPolicy() { } - public SharedStoreMasterPolicy(long failbackDelay, boolean failoverOnServerShutdown) { - this.failbackDelay = failbackDelay; + public SharedStoreMasterPolicy(boolean failoverOnServerShutdown) { this.failoverOnServerShutdown = failoverOnServerShutdown; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public void setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; } public boolean isFailoverOnServerShutdown() { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreSlavePolicy.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreSlavePolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreSlavePolicy.java index 7f2693c..af6a955 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreSlavePolicy.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ha/SharedStoreSlavePolicy.java @@ -25,8 +25,6 @@ import java.util.Map; public class SharedStoreSlavePolicy extends BackupPolicy { - private long failbackDelay = ActiveMQDefaultConfiguration.getDefaultFailbackDelay(); - private boolean failoverOnServerShutdown = ActiveMQDefaultConfiguration.isDefaultFailoverOnServerShutdown(); private boolean allowAutoFailBack = ActiveMQDefaultConfiguration.isDefaultAllowAutoFailback(); @@ -37,24 +35,23 @@ public class SharedStoreSlavePolicy extends BackupPolicy { public SharedStoreSlavePolicy() { } - public SharedStoreSlavePolicy(long failbackDelay, - boolean failoverOnServerShutdown, + public SharedStoreSlavePolicy(boolean failoverOnServerShutdown, boolean restartBackup, boolean allowAutoFailBack, ScaleDownPolicy scaleDownPolicy) { - this.failbackDelay = failbackDelay; this.failoverOnServerShutdown = failoverOnServerShutdown; this.restartBackup = restartBackup; this.allowAutoFailBack = allowAutoFailBack; this.scaleDownPolicy = scaleDownPolicy; } + @Deprecated public long getFailbackDelay() { - return failbackDelay; + return -1; } + @Deprecated public void setFailbackDelay(long failbackDelay) { - this.failbackDelay = failbackDelay; } public boolean isFailoverOnServerShutdown() { @@ -67,7 +64,7 @@ public class SharedStoreSlavePolicy extends BackupPolicy { public SharedStoreMasterPolicy getSharedStoreMasterPolicy() { if (sharedStoreMasterPolicy == null) { - sharedStoreMasterPolicy = new SharedStoreMasterPolicy(failbackDelay, failoverOnServerShutdown); + sharedStoreMasterPolicy = new SharedStoreMasterPolicy(failoverOnServerShutdown); } return sharedStoreMasterPolicy; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/FileLockNodeManager.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/FileLockNodeManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/FileLockNodeManager.java index acb431d..6169cef 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/FileLockNodeManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/FileLockNodeManager.java @@ -23,6 +23,7 @@ import java.nio.channels.FileLock; import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException; import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.server.ActivateCallback; import org.apache.activemq.artemis.core.server.ActiveMQServerLogger; import org.apache.activemq.artemis.core.server.NodeManager; import org.apache.activemq.artemis.utils.UUID; @@ -153,7 +154,7 @@ public class FileLockNodeManager extends NodeManager { } @Override - public void startLiveNode() throws Exception { + public ActivateCallback startLiveNode() throws Exception { setFailingBack(); String timeoutMessage = lockAcquisitionTimeout == -1 ? "indefinitely" : lockAcquisitionTimeout + " milliseconds"; @@ -164,7 +165,29 @@ public class FileLockNodeManager extends NodeManager { ActiveMQServerLogger.LOGGER.obtainedLiveLock(); - setLive(); + return new ActivateCallback() { + @Override + public void preActivate() { + } + + @Override + public void activated() { + } + + @Override + public void deActivate() { + } + + @Override + public void activationComplete() { + try { + setLive(); + } + catch (Exception e) { + e.printStackTrace(); + } + } + }; } @Override @@ -183,6 +206,13 @@ public class FileLockNodeManager extends NodeManager { } } + @Override + public void awaitLiveStatus() throws Exception { + while (getState() != LIVE) { + Thread.sleep(2000); + } + } + private void setLive() throws Exception { writeFileLockStatus(FileLockNodeManager.LIVE); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/InVMNodeManager.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/InVMNodeManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/InVMNodeManager.java index 726cb50..48f1627 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/InVMNodeManager.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/InVMNodeManager.java @@ -22,6 +22,7 @@ import java.util.concurrent.Semaphore; import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException; import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.server.ActivateCallback; import org.apache.activemq.artemis.core.server.NodeManager; import org.apache.activemq.artemis.utils.UUIDGenerator; @@ -91,15 +92,44 @@ public final class InVMNodeManager extends NodeManager { } @Override + public void awaitLiveStatus() throws Exception { + while (state != LIVE) { + Thread.sleep(10); + } + } + + @Override public void startBackup() throws Exception { backupLock.acquire(); } @Override - public void startLiveNode() throws Exception { + public ActivateCallback startLiveNode() throws Exception { state = FAILING_BACK; liveLock.acquire(); - state = LIVE; + return new ActivateCallback() { + @Override + public void preActivate() { + } + + @Override + public void activated() { + } + + @Override + public void deActivate() { + } + + @Override + public void activationComplete() { + try { + state = LIVE; + } + catch (Exception e) { + e.printStackTrace(); + } + } + }; } @Override @@ -110,8 +140,6 @@ public final class InVMNodeManager extends NodeManager { @Override public void crashLiveServer() throws Exception { - //overkill as already set to live - state = LIVE; liveLock.release(); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java index 50c45e6..52d6260 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java @@ -157,7 +157,7 @@ public class SharedNothingLiveActivation extends LiveActivation { Thread t = new Thread(new Runnable() { public void run() { try { - activeMQServer.getStorageManager().startReplication(replicationManager, activeMQServer.getPagingManager(), activeMQServer.getNodeID().toString(), isFailBackRequest && replicatedPolicy.isAllowAutoFailBack()); + activeMQServer.getStorageManager().startReplication(replicationManager, activeMQServer.getPagingManager(), activeMQServer.getNodeID().toString(), isFailBackRequest && replicatedPolicy.isAllowAutoFailBack(), replicatedPolicy.getInitialReplicationSyncTimeout()); clusterConnection.nodeAnnounced(System.currentTimeMillis(), activeMQServer.getNodeID().toString(), replicatedPolicy.getGroupName(), replicatedPolicy.getScaleDownGroupName(), pair, true); @@ -168,13 +168,7 @@ public class SharedNothingLiveActivation extends LiveActivation { BackupTopologyListener listener1 = new BackupTopologyListener(activeMQServer.getNodeID().toString()); clusterConnection.addClusterTopologyListener(listener1); if (listener1.waitForBackup()) { - try { - Thread.sleep(replicatedPolicy.getFailbackDelay()); - } - catch (InterruptedException e) { - // - } - //if we have to many backups kept or arent configured to restart just stop, otherwise restart as a backup + //if we have to many backups kept or are not configured to restart just stop, otherwise restart as a backup if (!replicatedPolicy.getReplicaPolicy().isRestartBackup() && activeMQServer.countNumberOfCopiedJournals() >= replicatedPolicy.getReplicaPolicy().getMaxSavedReplicatedJournalsSize() && replicatedPolicy.getReplicaPolicy().getMaxSavedReplicatedJournalsSize() >= 0) { activeMQServer.stop(true); ActiveMQServerLogger.LOGGER.stopReplicatedBackupAfterFailback(); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreBackupActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreBackupActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreBackupActivation.java index e556b5d..0aee108 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreBackupActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreBackupActivation.java @@ -16,6 +16,9 @@ */ package org.apache.activemq.artemis.core.server.impl; +import java.nio.channels.ClosedChannelException; +import java.util.concurrent.TimeUnit; + import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.core.config.Configuration; import org.apache.activemq.artemis.core.paging.PagingManager; @@ -30,9 +33,6 @@ import org.apache.activemq.artemis.core.server.cluster.ha.SharedStoreSlavePolicy import org.apache.activemq.artemis.core.server.group.GroupingHandler; import org.apache.activemq.artemis.core.server.management.ManagementService; -import java.nio.channels.ClosedChannelException; -import java.util.concurrent.TimeUnit; - public final class SharedStoreBackupActivation extends Activation { //this is how we act as a backup @@ -191,38 +191,51 @@ public final class SharedStoreBackupActivation extends Activation { } private class FailbackChecker implements Runnable { + BackupTopologyListener backupListener; + + FailbackChecker() { + backupListener = new BackupTopologyListener(activeMQServer.getNodeID().toString()); + activeMQServer.getClusterManager().getDefaultConnection(null).addClusterTopologyListener(backupListener); + } private boolean restarting = false; public void run() { try { if (!restarting && activeMQServer.getNodeManager().isAwaitingFailback()) { - ActiveMQServerLogger.LOGGER.awaitFailBack(); - restarting = true; - Thread t = new Thread(new Runnable() { - public void run() { - try { - ActiveMQServerLogger.LOGGER.debug(activeMQServer + "::Stopping live node in favor of failback"); - - activeMQServer.stop(true, false, true); - // We need to wait some time before we start the backup again - // otherwise we may eventually start before the live had a chance to get it - Thread.sleep(sharedStoreSlavePolicy.getFailbackDelay()); - synchronized (failbackCheckerGuard) { - if (cancelFailBackChecker || !sharedStoreSlavePolicy.isRestartBackup()) - return; - - activeMQServer.setHAPolicy(sharedStoreSlavePolicy); - ActiveMQServerLogger.LOGGER.debug(activeMQServer + "::Starting backup node now after failback"); - activeMQServer.start(); + if (backupListener.waitForBackup()) { + ActiveMQServerLogger.LOGGER.awaitFailBack(); + restarting = true; + Thread t = new Thread(new Runnable() { + public void run() { + try { + ActiveMQServerLogger.LOGGER.debug(activeMQServer + "::Stopping live node in favor of failback"); + + NodeManager nodeManager = activeMQServer.getNodeManager(); + activeMQServer.stop(true, false, true); + + // ensure that the server to which we are failing back actually starts fully before we restart + nodeManager.start(); + nodeManager.awaitLiveStatus(); + nodeManager.stop(); + + synchronized (failbackCheckerGuard) { + if (cancelFailBackChecker || !sharedStoreSlavePolicy.isRestartBackup()) + return; + + activeMQServer.setHAPolicy(sharedStoreSlavePolicy); + ActiveMQServerLogger.LOGGER.debug(activeMQServer + "::Starting backup node now after failback"); + activeMQServer.start(); + } + } + catch (Exception e) { + ActiveMQServerLogger.LOGGER.serverRestartWarning(); + e.printStackTrace(); } } - catch (Exception e) { - ActiveMQServerLogger.LOGGER.serverRestartWarning(); - } - } - }); - t.start(); + }); + t.start(); + } } } catch (Exception e) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreLiveActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreLiveActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreLiveActivation.java index 95c24ab..f48bb6c 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreLiveActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedStoreLiveActivation.java @@ -55,10 +55,9 @@ public final class SharedStoreLiveActivation extends LiveActivation { } activeMQServer.getBackupManager().start(); activeMQServer.getBackupManager().announceBackup(); - Thread.sleep(sharedStoreMasterPolicy.getFailbackDelay()); } - activeMQServer.getNodeManager().startLiveNode(); + activeMQServer.registerActivateCallback(activeMQServer.getNodeManager().startLiveNode()); if (activeMQServer.getState() == ActiveMQServerImpl.SERVER_STATE.STOPPED || activeMQServer.getState() == ActiveMQServerImpl.SERVER_STATE.STOPPING) { return; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/main/resources/schema/artemis-configuration.xsd ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd index 6f25831..5ab8b76 100644 --- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd +++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd @@ -1624,6 +1624,14 @@ </xsd:documentation> </xsd:annotation> </xsd:element> + <xsd:element name="initial-replication-sync-timeout" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0"> + <xsd:annotation> + <xsd:documentation> + The amount of time to wait for the replica to acknowledge it has received all the necessary data from + the replicating server at the final step of the initial replication synchronization process. + </xsd:documentation> + </xsd:annotation> + </xsd:element> </xsd:all> </xsd:complexType> <xsd:complexType name="replicaPolicyType"> @@ -1681,7 +1689,16 @@ <xsd:element name="failback-delay" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0"> <xsd:annotation> <xsd:documentation> - if we have to start as a replicated server this is the delay to wait before fail-back occurs + DEPRECATED: if we have to start as a replicated server this is the delay to wait before fail-back occurs + </xsd:documentation> + </xsd:annotation> + </xsd:element> + <xsd:element name="initial-replication-sync-timeout" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0"> + <xsd:annotation> + <xsd:documentation> + If we have to start as a replicated server this is the amount of time to wait for the replica to + acknowledge it has received all the necessary data from the replicating server at the final step + of the initial replication synchronization process. </xsd:documentation> </xsd:annotation> </xsd:element> @@ -1736,7 +1753,7 @@ <xsd:element name="failback-delay" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0"> <xsd:annotation> <xsd:documentation> - delay to wait before fail-back occurs on (live's) restart + DEPRECATED: delay to wait before fail-back occurs on (live's) restart </xsd:documentation> </xsd:annotation> </xsd:element> @@ -1764,7 +1781,7 @@ <xsd:element name="failback-delay" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0"> <xsd:annotation> <xsd:documentation> - delay to wait before fail-back occurs on (live's) restart + DEPRECATED: delay to wait before fail-back occurs on (live's) restart </xsd:documentation> </xsd:annotation> </xsd:element> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/HAPolicyConfigurationTest.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/HAPolicyConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/HAPolicyConfigurationTest.java index 9f5d2c5..7fd25f1 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/HAPolicyConfigurationTest.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/HAPolicyConfigurationTest.java @@ -121,6 +121,7 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertEquals(replicatedPolicy.getGroupName(), "purple"); assertTrue(replicatedPolicy.isCheckForLiveServer()); assertEquals(replicatedPolicy.getClusterName(), "abcdefg"); + assertEquals(replicatedPolicy.getInitialReplicationSyncTimeout(), 9876); } finally { server.stop(); @@ -142,6 +143,8 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertEquals(replicaPolicy.getMaxSavedReplicatedJournalsSize(), 22); assertEquals(replicaPolicy.getClusterName(), "33rrrrr"); assertFalse(replicaPolicy.isRestartBackup()); + assertTrue(replicaPolicy.isAllowFailback()); + assertEquals(replicaPolicy.getInitialReplicationSyncTimeout(), 9876); ScaleDownPolicy scaleDownPolicy = replicaPolicy.getScaleDownPolicy(); assertNotNull(scaleDownPolicy); assertEquals(scaleDownPolicy.getGroupName(), "boo!"); @@ -219,7 +222,6 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { HAPolicy haPolicy = server.getHAPolicy(); assertTrue(haPolicy instanceof SharedStoreMasterPolicy); SharedStoreMasterPolicy masterPolicy = (SharedStoreMasterPolicy) haPolicy; - assertEquals(masterPolicy.getFailbackDelay(), 3456); assertFalse(masterPolicy.isFailoverOnServerShutdown()); } finally { @@ -237,11 +239,10 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertTrue(activation instanceof SharedStoreBackupActivation); HAPolicy haPolicy = server.getHAPolicy(); assertTrue(haPolicy instanceof SharedStoreSlavePolicy); - SharedStoreSlavePolicy replicaPolicy = (SharedStoreSlavePolicy) haPolicy; - assertEquals(replicaPolicy.getFailbackDelay(), 9876); - assertFalse(replicaPolicy.isFailoverOnServerShutdown()); - assertFalse(replicaPolicy.isRestartBackup()); - ScaleDownPolicy scaleDownPolicy = replicaPolicy.getScaleDownPolicy(); + SharedStoreSlavePolicy sharedStoreSlavePolicy = (SharedStoreSlavePolicy) haPolicy; + assertFalse(sharedStoreSlavePolicy.isFailoverOnServerShutdown()); + assertFalse(sharedStoreSlavePolicy.isRestartBackup()); + ScaleDownPolicy scaleDownPolicy = sharedStoreSlavePolicy.getScaleDownPolicy(); assertNotNull(scaleDownPolicy); assertEquals(scaleDownPolicy.getGroupName(), "boo!"); assertEquals(scaleDownPolicy.getDiscoveryGroup(), "wahey"); @@ -264,11 +265,10 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertTrue(activation instanceof SharedStoreBackupActivation); HAPolicy haPolicy = server.getHAPolicy(); assertTrue(haPolicy instanceof SharedStoreSlavePolicy); - SharedStoreSlavePolicy replicaPolicy = (SharedStoreSlavePolicy) haPolicy; - assertEquals(replicaPolicy.getFailbackDelay(), 5678); - assertTrue(replicaPolicy.isFailoverOnServerShutdown()); - assertTrue(replicaPolicy.isRestartBackup()); - ScaleDownPolicy scaleDownPolicy = replicaPolicy.getScaleDownPolicy(); + SharedStoreSlavePolicy sharedStoreSlavePolicy = (SharedStoreSlavePolicy) haPolicy; + assertTrue(sharedStoreSlavePolicy.isFailoverOnServerShutdown()); + assertTrue(sharedStoreSlavePolicy.isRestartBackup()); + ScaleDownPolicy scaleDownPolicy = sharedStoreSlavePolicy.getScaleDownPolicy(); assertNotNull(scaleDownPolicy); assertEquals(scaleDownPolicy.getGroupName(), "boo!"); assertEquals(scaleDownPolicy.getDiscoveryGroup(), null); @@ -293,11 +293,10 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertTrue(activation instanceof SharedStoreBackupActivation); HAPolicy haPolicy = server.getHAPolicy(); assertTrue(haPolicy instanceof SharedStoreSlavePolicy); - SharedStoreSlavePolicy replicaPolicy = (SharedStoreSlavePolicy) haPolicy; - assertEquals(replicaPolicy.getFailbackDelay(), 5678); - assertTrue(replicaPolicy.isFailoverOnServerShutdown()); - assertTrue(replicaPolicy.isRestartBackup()); - ScaleDownPolicy scaleDownPolicy = replicaPolicy.getScaleDownPolicy(); + SharedStoreSlavePolicy sharedStoreSlavePolicy = (SharedStoreSlavePolicy) haPolicy; + assertTrue(sharedStoreSlavePolicy.isFailoverOnServerShutdown()); + assertTrue(sharedStoreSlavePolicy.isRestartBackup()); + ScaleDownPolicy scaleDownPolicy = sharedStoreSlavePolicy.getScaleDownPolicy(); assertNull(scaleDownPolicy); } finally { @@ -349,10 +348,8 @@ public class HAPolicyConfigurationTest extends ActiveMQTestBase { assertNotNull(livePolicy); assertFalse(livePolicy.isFailoverOnServerShutdown()); - assertEquals(livePolicy.getFailbackDelay(), 1234); SharedStoreSlavePolicy backupPolicy = (SharedStoreSlavePolicy) colocatedPolicy.getBackupPolicy(); assertNotNull(backupPolicy); - assertEquals(backupPolicy.getFailbackDelay(), 44); assertFalse(backupPolicy.isFailoverOnServerShutdown()); assertFalse(backupPolicy.isRestartBackup()); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/colocated-hapolicy-config2.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/colocated-hapolicy-config2.xml b/artemis-server/src/test/resources/colocated-hapolicy-config2.xml index ff65b54..ca13b65 100644 --- a/artemis-server/src/test/resources/colocated-hapolicy-config2.xml +++ b/artemis-server/src/test/resources/colocated-hapolicy-config2.xml @@ -28,19 +28,15 @@ <request-backup>false</request-backup> <backup-port-offset>33</backup-port-offset> <master> - <failback-delay>1234</failback-delay> <failover-on-shutdown>false</failover-on-shutdown> </master> <slave> - <failback-delay>44</failback-delay> <failover-on-shutdown>false</failover-on-shutdown> <restart-backup>false</restart-backup> <scale-down/> </slave> </colocated> - </shared-store> </ha-policy> - </core>c - + </core> </configuration> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/replica-hapolicy-config.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/replica-hapolicy-config.xml b/artemis-server/src/test/resources/replica-hapolicy-config.xml index 7183673..03983fc 100644 --- a/artemis-server/src/test/resources/replica-hapolicy-config.xml +++ b/artemis-server/src/test/resources/replica-hapolicy-config.xml @@ -30,7 +30,7 @@ <cluster-name>33rrrrr</cluster-name> <restart-backup>false</restart-backup> <allow-failback>true</allow-failback> - <failback-delay>9876</failback-delay> + <initial-replication-sync-timeout>9876</initial-replication-sync-timeout> <scale-down> <!--a grouping of servers that can be scaled down to--> <group-name>boo!</group-name> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/replicated-hapolicy-config.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/replicated-hapolicy-config.xml b/artemis-server/src/test/resources/replicated-hapolicy-config.xml index d419314..8195e45 100644 --- a/artemis-server/src/test/resources/replicated-hapolicy-config.xml +++ b/artemis-server/src/test/resources/replicated-hapolicy-config.xml @@ -26,6 +26,7 @@ <group-name>purple</group-name> <check-for-live-server>true</check-for-live-server> <cluster-name>abcdefg</cluster-name> + <initial-replication-sync-timeout>9876</initial-replication-sync-timeout> </master> </replication> </ha-policy> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/shared-store-master-hapolicy-config.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/shared-store-master-hapolicy-config.xml b/artemis-server/src/test/resources/shared-store-master-hapolicy-config.xml index cb55b42..132cd00 100644 --- a/artemis-server/src/test/resources/shared-store-master-hapolicy-config.xml +++ b/artemis-server/src/test/resources/shared-store-master-hapolicy-config.xml @@ -22,11 +22,9 @@ <ha-policy> <shared-store> <master> - <failback-delay>3456</failback-delay> <failover-on-shutdown>false</failover-on-shutdown> </master> </shared-store> </ha-policy> </core> - </configuration> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/shared-store-slave-hapolicy-config.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config.xml b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config.xml index 57dff39..28c6051 100644 --- a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config.xml +++ b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config.xml @@ -26,7 +26,6 @@ <shared-store> <slave> <allow-failback>true</allow-failback> - <failback-delay>9876</failback-delay> <failover-on-shutdown>false</failover-on-shutdown> <restart-backup>false</restart-backup> <scale-down> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/shared-store-slave-hapolicy-config2.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config2.xml b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config2.xml index 3c1c19b..57acd13 100644 --- a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config2.xml +++ b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config2.xml @@ -22,7 +22,6 @@ <ha-policy> <shared-store> <slave> - <failback-delay>5678</failback-delay> <failover-on-shutdown>true</failover-on-shutdown> <restart-backup>true</restart-backup> <scale-down> @@ -38,5 +37,4 @@ </shared-store> </ha-policy> </core> - </configuration> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ef5a9809/artemis-server/src/test/resources/shared-store-slave-hapolicy-config3.xml ---------------------------------------------------------------------- diff --git a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config3.xml b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config3.xml index 1c62ae3..991ea55 100644 --- a/artemis-server/src/test/resources/shared-store-slave-hapolicy-config3.xml +++ b/artemis-server/src/test/resources/shared-store-slave-hapolicy-config3.xml @@ -22,12 +22,10 @@ <ha-policy> <shared-store> <slave> - <failback-delay>5678</failback-delay> <failover-on-shutdown>true</failover-on-shutdown> <restart-backup>true</restart-backup> </slave> </shared-store> </ha-policy> </core> - </configuration>
