9:42 a.m.
Author: borges
Date: 2012-02-21 10:42:48 -0500 (Tue, 21 Feb 2012)
New Revision: 12156
Added:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/LiveIsStoppingMessage.java
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketDecoder.java
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketImpl.java
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/QuorumManager.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/QuorumFailOverTest.java
Log:
HORNETQ-720 Backup should not worry about split brain if live had a clean exit.
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketDecoder.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -97,6 +97,7 @@
import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage;
import org.hornetq.core.protocol.core.impl.wireformat.DisconnectMessage;
import org.hornetq.core.protocol.core.impl.wireformat.HornetQExceptionMessage;
+import org.hornetq.core.protocol.core.impl.wireformat.LiveIsStoppingMessage;
import org.hornetq.core.protocol.core.impl.wireformat.NodeAnnounceMessage;
import org.hornetq.core.protocol.core.impl.wireformat.NullResponseMessage;
import org.hornetq.core.protocol.core.impl.wireformat.PacketsConfirmedMessage;
@@ -563,6 +564,11 @@
packet = new ReplicationSyncFileMessage();
break;
}
+ case PacketImpl.REPLICATION_SCHEDULED_FAILOVER:
+ {
+ packet = new LiveIsStoppingMessage();
+ break;
+ }
default:
{
throw new IllegalArgumentException("Invalid type: " + packetType);
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketImpl.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/PacketImpl.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -204,6 +204,7 @@
public static final byte BACKUP_REGISTRATION_FAILED = 116;
public static final byte REPLICATION_START_FINISH_SYNC = 120;
+ public static final byte REPLICATION_SCHEDULED_FAILOVER = 121;
// Static --------------------------------------------------------
Added:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/LiveIsStoppingMessage.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/LiveIsStoppingMessage.java
(rev 0)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/LiveIsStoppingMessage.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -0,0 +1,21 @@
+/**
+ *
+ */
+package org.hornetq.core.protocol.core.impl.wireformat;
+
+import org.hornetq.core.protocol.core.impl.PacketImpl;
+
+/**
+ * Message indicating that the live is stopping.
+ * <p>
+ * The backup starts the fail-over immediately after receiving this.
+ */
+public final class LiveIsStoppingMessage extends PacketImpl
+{
+
+ public LiveIsStoppingMessage()
+ {
+ super(PacketImpl.REPLICATION_SCHEDULED_FAILOVER);
+ }
+
+}
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -119,9 +119,9 @@
private boolean started;
private QuorumManager quorumManager;
-
+
//https://community.jboss.org/thread/195519
- private Object stopLock = new Object();
+ private final Object stopLock = new Object();
// Constructors --------------------------------------------------
public ReplicationEndpoint(final HornetQServerImpl server, IOCriticalErrorListener
criticalErrorListener)
@@ -165,7 +165,7 @@
{
return;
}
-
+
if (type == PacketImpl.REPLICATION_APPEND)
{
handleAppendAddRecord((ReplicationAddMessage) packet);
@@ -223,10 +223,13 @@
{
handleReplicationSynchronization((ReplicationSyncFileMessage) packet);
}
- else
- {
- log.warn("Packet " + packet
- + " can't be processed by the ReplicationEndpoint");
+ else if (type == PacketImpl.REPLICATION_SCHEDULED_FAILOVER)
+ {
+ handleLiveStopping();
+ }
+ else
+ {
+ log.warn("Packet " + packet + " can't be processed by the
ReplicationEndpoint");
}
}
}
@@ -246,6 +249,14 @@
channel.send(response);
}
+ /**
+ * @throws HornetQException
+ */
+ private void handleLiveStopping() throws HornetQException
+ {
+ server.remoteFailOver();
+ }
+
public boolean isStarted()
{
return started;
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -135,4 +135,12 @@
* @param largeMessageIDs
*/
void sendLargeMessageIdListMessage(List<Long> largeMessageIDs);
+
+ /**
+ * Notifies the backup that the live server is stopping.
+ * <p>
+ * This notification allows the backup to skip quorum voting (or any other measure to
avoid
+ * 'split-brain') and do a faster fail-over.
+ */
+ void sendLiveIsStopping();
}
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -41,6 +41,7 @@
import org.hornetq.core.protocol.core.Packet;
import org.hornetq.core.protocol.core.impl.ChannelImpl.CHANNEL_ID;
import org.hornetq.core.protocol.core.impl.PacketImpl;
+import org.hornetq.core.protocol.core.impl.wireformat.LiveIsStoppingMessage;
import org.hornetq.core.protocol.core.impl.wireformat.ReplicationAddMessage;
import org.hornetq.core.protocol.core.impl.wireformat.ReplicationAddTXMessage;
import org.hornetq.core.protocol.core.impl.wireformat.ReplicationCommitMessage;
@@ -314,7 +315,7 @@
synchronized (replicationLock)
{
- enabled = false;
+ enabled = false;
// Complete any pending operations...
// Case the backup crashed, this should clean up any pending requests
@@ -601,4 +602,13 @@
sendReplicatePacket(new ReplicationStartSyncMessage(largeMessageIDs));
}
+
+ /**
+ * Notifies the backup that the live is about to stop.
+ */
+ public void sendLiveIsStopping()
+ {
+ if (enabled)
+ sendReplicatePacket(new LiveIsStoppingMessage());
+ }
}
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -260,7 +260,7 @@
private Thread backupActivationThread;
private Activation activation;
-
+
private final ShutdownOnCriticalErrorListener shutdownOnCriticalIO = new
ShutdownOnCriticalErrorListener();
// Constructors
@@ -487,7 +487,7 @@
{
stop(failoverOnServerShutdown, false);
}
-
+
private void stop(boolean failoverOnServerShutdown, boolean criticalIOError) throws
Exception
{
synchronized (this)
@@ -497,6 +497,10 @@
return;
}
+ if (replicationManager!=null) {
+ replicationManager.sendLiveIsStopping();
+ }
+
connectorsService.stop();
// we stop the groupingHandler before we stop the cluster manager so binding
mappings
@@ -539,7 +543,7 @@
log.warn(e.getMessage(), e);
}
}
-
+
storageManager.clearContext();
synchronized (this)
@@ -651,9 +655,9 @@
{
// Ignore
}
-
+
securityStore.stop();
-
+
threadPool = null;
scheduledPool = null;
@@ -679,7 +683,7 @@
initialised = new CountDownLatch(1);
}
}
-
+
// to display in the log message
SimpleString tempNodeID = getNodeID();
@@ -804,7 +808,7 @@
{
return started;
}
-
+
public boolean isStopped()
{
return stopped;
@@ -1048,15 +1052,15 @@
{
storageManager.deleteQueueBinding(queue.getID());
}
-
+
if (queue.getPageSubscription() != null)
{
queue.getPageSubscription().close();
}
-
+
PageSubscription subs = queue.getPageSubscription();
-
+
if (subs != null)
{
subs.cleanupEntries(true);
@@ -1241,8 +1245,8 @@
addressSettingsRepository);
}
- /**
- * This method is protected as it may be used as a hook for creating a custom storage
manager (on tests for instance)
+ /**
+ * This method is protected as it may be used as a hook for creating a custom storage
manager (on tests for instance)
*/
private StorageManager createStorageManager()
{
@@ -1742,8 +1746,8 @@
pageSubscription.close();
throw e;
}
-
+
managementService.registerAddress(address);
managementService.registerQueue(queue, address, storageManager);
@@ -2051,19 +2055,19 @@
}
}
}
-
+
private final class ShutdownOnCriticalErrorListener implements
IOCriticalErrorListener
{
boolean failedAlready = false;
-
+
public synchronized void onIOException(int code, String message, SequentialFile
file)
{
if (!failedAlready)
{
failedAlready = true;
-
+
log.warn("Critical IO Error, shutting down the server. code=" +
code + ", message=" + message);
-
+
new Thread()
{
@Override
@@ -2092,6 +2096,7 @@
{
private ServerLocatorInternal serverLocator0;
private volatile boolean failedConnection;
+ private volatile boolean failOver;
public void run()
{
@@ -2161,7 +2166,7 @@
"] started, waiting live to fail before it gets active");
started = true;
- // Server node (i.e. Life node) is not running, now the backup takes over.
+ // Server node (i.e. Live node) is not running, now the backup takes over.
// we must remember to close stuff we don't need any more
synchronized (quorumManager)
{
@@ -2170,11 +2175,10 @@
while (true)
{
quorumManager.wait();
- break;
-// if (!started || quorumManager.isNodeDown())
-// {
-// break;
-// }
+ if (failOver || !started || quorumManager.isNodeDown())
+ {
+ break;
+ }
}
}
@@ -2245,6 +2249,14 @@
nodeManager.stopBackup();
}
}
+
+ /**
+ * Live has notified this server that it is going to stop.
+ */
+ public void failOver()
+ {
+ failOver = true;
+ }
}
@@ -2285,7 +2297,7 @@
}
}
}
-
+
/** This seems duplicate code all over the place, but for security reasons we
can't let something like this to be open in a
* utility class, as it would be a door to load anything you like in a safe VM.
* For that reason any class trying to do a privileged block should do with the
AccessController directly.
@@ -2359,10 +2371,8 @@
{
throw (HornetQException)e;
}
- else
- {
- throw new HornetQException(HornetQException.INTERNAL_ERROR, "Error
trying to start replication", e);
- }
+
+ throw new HornetQException(HornetQException.INTERNAL_ERROR, "Error
trying to start replication", e);
}
}
}
@@ -2408,4 +2418,21 @@
}
}
+ /**
+ * @throws HornetQException
+ */
+ public void remoteFailOver() throws HornetQException
+ {
+ if (!configuration.isBackup() || configuration.isSharedStore())
+ {
+ throw new HornetQException(HornetQException.INTERNAL_ERROR);
+ }
+ if (!backupUpToDate) return;
+ if (activation instanceof SharedNothingBackupActivation)
+ {
+ ((SharedNothingBackupActivation)activation).failOver();
+ }
+
+ }
+
}
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/QuorumManager.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/QuorumManager.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/QuorumManager.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -8,6 +8,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.hornetq.api.core.Pair;
@@ -28,15 +29,16 @@
public final class QuorumManager implements ClusterTopologyListener
{
- // private static final Logger LOG = Logger.getLogger(QuorumManager.class);
-
// volatile boolean started;
private final ServerLocator locator;
private String targetServerID = "";
private final Map<String, Pair<TransportConfiguration,
TransportConfiguration>> nodes =
new ConcurrentHashMap<String, Pair<TransportConfiguration,
TransportConfiguration>>();
- private static final long DISCOVERY_TIMEOUT = 3;
+ /** safety parameter to make _sure_ we get out of await() */
+ private static final int LATCH_TIMEOUT = 60;
+ private static final long DISCOVERY_TIMEOUT = 5;
+
public QuorumManager(ServerLocator serverLocator)
{
this.locator = serverLocator;
@@ -75,12 +77,11 @@
public boolean isNodeDown()
{
- boolean liveShutdownCleanly = !nodes.containsKey(targetServerID);
- boolean noOtherServersAround = nodes.size() == 0;
- if (liveShutdownCleanly || noOtherServersAround)
+ if (nodes.size() == 0)
+ {
return true;
+ }
// go for the vote...
- // Set<ServerLocator> currentNodes = new HashSet(nodes.entrySet());
final int size = nodes.size();
Set<ServerLocator> locatorsList = new HashSet<ServerLocator>(size);
AtomicInteger pingCount = new AtomicInteger(0);
@@ -104,7 +105,7 @@
}
try
{
- latch.await();
+ latch.await(LATCH_TIMEOUT, TimeUnit.SECONDS);
}
catch (InterruptedException interruption)
{
@@ -163,8 +164,8 @@
finally
{
latch.countDown();
+ locator.close();
}
}
-
}
}
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
===================================================================
---
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -251,6 +251,4 @@
{
return TransportConfigurationUtils.getInVMConnector(live);
}
-
-
}
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/QuorumFailOverTest.java
===================================================================
---
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/QuorumFailOverTest.java 2012-02-21
14:06:10 UTC (rev 12155)
+++
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/QuorumFailOverTest.java 2012-02-21
15:42:48 UTC (rev 12156)
@@ -2,6 +2,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
import org.hornetq.api.core.Pair;
import org.hornetq.api.core.TransportConfiguration;
@@ -12,27 +13,37 @@
public void testQuorumVoting() throws Exception
{
+
setupCluster();
+
startServers(0, 1, 2, 3, 4, 5);
+ for (int i : new int[] { 0, 1, 2 })
+ {
+ setupSessionFactory(i, i + 3, isNetty(), false);
+ }
+ createQueue(0, QUEUES_TESTADDRESS, QUEUE_NAME, null, true);
+
final TopologyListener liveTopologyListener = new
TopologyListener("LIVE-1");
- fail("must rewrite to use new interfaces");
- // servers[0].getClusterManager().addClusterTopologyListener(liveTopologyListener,
true);
- final TopologyListener backupTopologyListener = new
TopologyListener("BACKUP-3");
- //
servers[3].getClusterManager().addClusterTopologyListener(backupTopologyListener, true);
+ locators[0].addClusterTopologyListener(liveTopologyListener);
+ final TopologyListener backupTopologyListener = new
TopologyListener("LIVE-2");
+ locators[1].addClusterTopologyListener(backupTopologyListener);
+
assertTrue("we assume 3 is a backup",
servers[3].getConfiguration().isBackup());
assertFalse("no shared storage",
servers[3].getConfiguration().isSharedStore());
- setupSessionFactory(0, 3, isNetty(), false);
- setupSessionFactory(1, 4, isNetty(), false);
- setupSessionFactory(2, 5, isNetty(), false);
+ // assertEquals(liveTopologyListener.toString(), 6,
liveTopologyListener.nodes.size());
+ // assertEquals(backupTopologyListener.toString(), 6,
backupTopologyListener.nodes.size());
- assertEquals(liveTopologyListener.toString(), 6,
liveTopologyListener.nodes.size());
- assertEquals(backupTopologyListener.toString(), 6,
backupTopologyListener.nodes.size());
+ failNode(0);
+ waitForBindings(3, QUEUES_TESTADDRESS, 1, 1, true);
- failNode(0);
+ assertTrue(servers[3].waitForInitialization(10, TimeUnit.SECONDS));
+ assertFalse("3 should have failed over ",
servers[3].getConfiguration().isBackup());
+ servers[1].stop();
+ assertFalse("4 should have failed over ",
servers[4].getConfiguration().isBackup());
}
@Override
@@ -41,7 +52,6 @@
return false;
}
-
private static class TopologyListener implements ClusterTopologyListener
{
final String prefix;
@@ -53,12 +63,11 @@
}
@Override
- public
- void nodeUP(long eventUID, String nodeID,
- Pair<TransportConfiguration, TransportConfiguration> connectorPair,
boolean last)
+ public void nodeUP(long eventUID, String nodeID,
+ Pair<TransportConfiguration, TransportConfiguration>
connectorPair, boolean last)
{
nodes.put(nodeID, connectorPair);
- System.out.println(prefix + " UP: " + nodeID);
+ System.out.println(prefix + " UP: " + nodeID + "
connectPair=" + connectorPair);
}
@Override