8:42 a.m.
Author: borges
Date: 2011-11-03 09:42:42 -0400 (Thu, 03 Nov 2011)
New Revision: 11655
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.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
Log:
HORNETQ-720 Use ClusterTopologyListener to monitor the live's status in the remote
replication case
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 2011-11-03
13:42:27 UTC (rev 11654)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java 2011-11-03
13:42:42 UTC (rev 11655)
@@ -19,6 +19,7 @@
import org.hornetq.core.protocol.core.Channel;
import org.hornetq.core.protocol.core.ChannelHandler;
import org.hornetq.core.server.HornetQComponent;
+import org.hornetq.core.server.impl.QuorumManager;
/**
* A ReplicationEndpoint
@@ -38,4 +39,11 @@
void registerJournal(final byte id, final Journal journal);
+ /**
+ * Sets the quorumManager used by the server in the replicationEndpoint. It is used to
inform the
+ * backup server of the live's nodeID.
+ * @param quorumManager
+ */
+ void setQuorumManager(QuorumManager quorumManager);
+
}
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-11-03
13:42:27 UTC (rev 11654)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-11-03
13:42:42 UTC (rev 11655)
@@ -68,6 +68,7 @@
import org.hornetq.core.server.LargeServerMessage;
import org.hornetq.core.server.ServerMessage;
import org.hornetq.core.server.impl.HornetQServerImpl;
+import org.hornetq.core.server.impl.QuorumManager;
/**
*
@@ -119,6 +120,8 @@
private boolean deletePages = true;
private boolean started;
+ private QuorumManager quorumManager;
+
// Constructors --------------------------------------------------
public ReplicationEndpointImpl(final HornetQServerImpl server, IOCriticalErrorListener
criticalErrorListener)
{
@@ -416,7 +419,7 @@
// Private -------------------------------------------------------
- private void finishSynchronization(String nodeID) throws Exception
+ private void finishSynchronization(String liveID) throws Exception
{
for (JournalContent jc : EnumSet.allOf(JournalContent.class))
{
@@ -477,7 +480,8 @@
}
}
journalsHolder = null;
- server.setRemoteBackupUpToDate(nodeID);
+ quorumManager.setLiveID(liveID);
+ server.setRemoteBackupUpToDate(liveID);
log.info("Backup server " + server + " is synchronized with
live-server.");
return;
}
@@ -903,4 +907,10 @@
return "JournalSyncFile(file=" + file.getAbsolutePath() +
")";
}
}
+
+ @Override
+ public void setQuorumManager(QuorumManager quorumManager)
+ {
+ this.quorumManager = quorumManager;
+ }
}
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 2011-11-03
13:42:27 UTC (rev 11654)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-03
13:42:42 UTC (rev 11655)
@@ -1959,11 +1959,8 @@
while (backupActivationThread.isAlive() && System.currentTimeMillis()
- start < timeout)
{
nodeManager.interrupt();
-
backupActivationThread.interrupt();
-
backupActivationThread.join(1000);
-
}
if (System.currentTimeMillis() - start >= timeout)
@@ -2049,6 +2046,7 @@
final TransportConfiguration config =
configuration.getConnectorConfigurations().get(liveConnectorName);
serverLocator =
(ServerLocatorInternal)HornetQClient.createServerLocatorWithHA(config);
final QuorumManager quorumManager = new QuorumManager(serverLocator);
+ replicationEndpoint.setQuorumManager(quorumManager);
serverLocator.setReconnectAttempts(-1);
@@ -2086,14 +2084,18 @@
// Server node (i.e. Life node) is not running, now the backup takes over.
// we must remember to close stuff we don't need any more
+ synchronized (quorumManager)
+ {
while (true)
{
- nodeManager.awaitLiveNode();
+ quorumManager.wait();
+ // nodeManager.awaitLiveNode();
break;
// if (!started || quorumManager.isNodeDown())
// {
// break;
// }
+ }
}
serverLocator.close();
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 2011-11-03
13:42:27 UTC (rev 11654)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/QuorumManager.java 2011-11-03
13:42:42 UTC (rev 11655)
@@ -25,14 +25,14 @@
* quorum will help a remote backup in deciding whether to replace its 'live'
server or to wait for
* it.
*/
-final class QuorumManager implements ClusterTopologyListener
+public final class QuorumManager implements ClusterTopologyListener
{
// private static final Logger LOG = Logger.getLogger(QuorumManager.class);
// volatile boolean started;
private final ServerLocator locator;
- private final String targetServerName = "";
+ private String targetServerID = "";
private final Map<String, Pair<TransportConfiguration,
TransportConfiguration>> nodes =
new ConcurrentHashMap<String, Pair<TransportConfiguration,
TransportConfiguration>>();
private static final long DISCOVERY_TIMEOUT = 3;
@@ -47,7 +47,7 @@
public void nodeUP(long eventUID, String nodeID, Pair<TransportConfiguration,
TransportConfiguration> connectorPair,
boolean last)
{
- if (targetServerName.equals(nodeID))
+ if (targetServerID.equals(nodeID))
{
return;
}
@@ -57,20 +57,25 @@
@Override
public void nodeDown(long eventUID, String nodeID)
{
- if (targetServerName.equals(nodeID))
+ if (targetServerID.equals(nodeID))
{
- // targetReturned = false;
- // trigger action
-
- // decide to wake backup
- locator.removeClusterTopologyListener(this);
+ if (!targetServerID.isEmpty())
+ synchronized (this)
+ {
+ notify();
+ }
}
nodes.remove(nodeID);
}
+ public void setLiveID(String liveID)
+ {
+ targetServerID = liveID;
+ }
+
public boolean isNodeDown()
{
- boolean liveShutdownCleanly = !nodes.containsKey(targetServerName);
+ boolean liveShutdownCleanly = !nodes.containsKey(targetServerID);
boolean noOtherServersAround = nodes.size() == 0;
if (liveShutdownCleanly || noOtherServersAround)
return true;
@@ -85,7 +90,7 @@
{
for (Entry<String, Pair<TransportConfiguration,
TransportConfiguration>> pair : nodes.entrySet())
{
- if (targetServerName.equals(pair.getKey()))
+ if (targetServerID.equals(pair.getKey()))
continue;
TransportConfiguration serverTC = pair.getValue().getA();
ServerLocatorImpl locator =
(ServerLocatorImpl)HornetQClient.createServerLocatorWithoutHA(serverTC);