You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by sl...@apache.org on 2013/01/31 02:27:29 UTC
[1/3] git commit: Add flapping detection logic in
ZkStateChangeListener. Each zkHelixManager will disconnect itself if flapping
is detected (become disconnected for X times in previous Y time window)
Add flapping detection logic in ZkStateChangeListener. Each
zkHelixManager will disconnect itself if flapping is detected (become
disconnected for X times in previous Y time window)
Project: http://git-wip-us.apache.org/repos/asf/incubator-helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-helix/commit/e690aff8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-helix/tree/e690aff8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-helix/diff/e690aff8
Branch: refs/heads/master
Commit: e690aff80eae7120a7392e430ce3b9e9c83a1a49
Parents: b67c181
Author: slu2011 <lu...@gmail.com>
Authored: Wed Jan 30 15:06:45 2013 -0800
Committer: slu2011 <lu...@gmail.com>
Committed: Wed Jan 30 15:06:45 2013 -0800
----------------------------------------------------------------------
.../apache/helix/manager/zk/ZKHelixManager.java | 64 ++++++++-
.../helix/manager/zk/ZkStateChangeListener.java | 45 ++++++-
.../test/java/org/apache/helix/ZkTestHelper.java | 39 +++++-
.../helix/manager/zk/TestZkDisconnectHistory.java | 109 +++++++++++++++
.../manager/zk/TestZkStateChangeListener.java | 80 +++++++++++
5 files changed, 328 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e690aff8/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index 7d51897..30a6137 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -117,6 +117,10 @@ public class ZKHelixManager implements HelixManager
List<PreConnectCallback> _preConnectCallbacks =
new LinkedList<PreConnectCallback>();
ZKPropertyTransferServer _transferServer = null;
+ int _flappingTimeWindowMs;
+ int _maxDisconnectThreshold;
+ public static final int FLAPPING_TIME_WINDIOW = 300000; // Default to 300 sec
+ public static final int MAX_DISCONNECT_THRESHOLD = 5;
public ZKHelixManager(String clusterName,
String instanceName,
@@ -126,10 +130,56 @@ public class ZKHelixManager implements HelixManager
logger.info("Create a zk-based cluster manager. clusterName:" + clusterName
+ ", instanceName:" + instanceName + ", type:" + instanceType + ", zkSvr:"
+ zkConnectString);
+ _flappingTimeWindowMs = FLAPPING_TIME_WINDIOW;
+ try
+ {
+ _flappingTimeWindowMs =
+ Integer.parseInt(System.getProperty("helixmanager.flappingTimeWindow", ""
+ + FLAPPING_TIME_WINDIOW));
+ }
+ catch (NumberFormatException e)
+ {
+ logger.warn("Exception while parsing helixmanager.flappingTimeWindow: "
+ + System.getProperty("helixmanager.flappingTimeWindow", "" + FLAPPING_TIME_WINDIOW));
+ }
+ _maxDisconnectThreshold = MAX_DISCONNECT_THRESHOLD;
+ try
+ {
+ _maxDisconnectThreshold =
+ Integer.parseInt(System.getProperty("helixmanager.maxDisconnectThreshold", ""
+ + MAX_DISCONNECT_THRESHOLD));
+ }
+ catch (NumberFormatException e)
+ {
+ logger.warn("Exception while parsing helixmanager.flappingTimeWindow: "
+ + System.getProperty("helixmanager.maxDisconnectThreshold", "" + MAX_DISCONNECT_THRESHOLD));
+ }
int sessionTimeoutInt = -1;
try
{
sessionTimeoutInt =
+ Integer.parseInt(System.getProperty("helixmanager.flappingTimeWindow", ""
+ + DEFAULT_SESSION_TIMEOUT));
+ }
+ catch (NumberFormatException e)
+ {
+ logger.warn("Exception while parsing session timeout: "
+ + System.getProperty("zk.session.timeout", "" + DEFAULT_SESSION_TIMEOUT));
+ }
+ try
+ {
+ sessionTimeoutInt =
+ Integer.parseInt(System.getProperty("zk.session.timeout", ""
+ + DEFAULT_SESSION_TIMEOUT));
+ }
+ catch (NumberFormatException e)
+ {
+ logger.warn("Exception while parsing session timeout: "
+ + System.getProperty("zk.session.timeout", "" + DEFAULT_SESSION_TIMEOUT));
+ }
+ try
+ {
+ sessionTimeoutInt =
Integer.parseInt(System.getProperty("zk.session.timeout", ""
+ DEFAULT_SESSION_TIMEOUT));
}
@@ -167,7 +217,7 @@ public class ZKHelixManager implements HelixManager
_instanceName = instanceName;
_instanceType = instanceType;
_zkConnectString = zkConnectString;
- _zkStateChangeListener = new ZkStateChangeListener(this);
+ _zkStateChangeListener = new ZkStateChangeListener(this, _flappingTimeWindowMs, _maxDisconnectThreshold);
_timer = null;
_handlers = new ArrayList<CallbackHandler>();
@@ -403,13 +453,18 @@ public class ZKHelixManager implements HelixManager
@Override
public void disconnect()
{
-
if (!isConnected())
{
- logger.warn("ClusterManager " + _instanceName + " already disconnected");
+ logger.error("ClusterManager " + _instanceName + " already disconnected");
return;
}
-
+ disconnectInternal();
+ }
+
+ void disconnectInternal()
+ {
+ // This function can be called when the connection are in bad state(e.g. flapping),
+ // in which isConnected() could be false and we want to disconnect from cluster.
logger.info("disconnect " + _instanceName + "(" + _instanceType + ") from "
+ _clusterName);
@@ -451,6 +506,7 @@ public class ZKHelixManager implements HelixManager
// HACK seems that zkClient is not sending DISCONNECT event
_zkStateChangeListener.disconnect();
logger.info("Cluster manager: " + _instanceName + " disconnected");
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e690aff8/helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java
index 2bf1242..834473e 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java
@@ -19,6 +19,9 @@ package org.apache.helix.manager.zk;
* under the License.
*/
+import java.util.LinkedList;
+import java.util.List;
+
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.log4j.Logger;
@@ -29,13 +32,23 @@ public class ZkStateChangeListener implements IZkStateListener
private volatile boolean _isConnected;
private volatile boolean _hasSessionExpired;
private final ZKHelixManager _zkHelixManager;
+
+ // Keep track of timestamps that zk State has become Disconnected
+ // If in a _timeWindowLengthMs window zk State has become Disconnected
+ // for more than_maxDisconnectThreshold times disconnect the zkHelixManager
+ List<Long> _disconnectTimeHistory = new LinkedList<Long>();
+ int _timeWindowLengthMs;
+ int _maxDisconnectThreshold;
private static Logger logger = Logger.getLogger(ZkStateChangeListener.class);
- public ZkStateChangeListener(ZKHelixManager zkHelixManager)
+ public ZkStateChangeListener(ZKHelixManager zkHelixManager, int timeWindowLengthMs, int maxDisconnectThreshold)
{
this._zkHelixManager = zkHelixManager;
-
+ _timeWindowLengthMs = timeWindowLengthMs;
+ // _maxDisconnectThreshold min value is 1.
+ // We don't want to disconnect from zk for the first time zkState become Disconnected
+ _maxDisconnectThreshold = maxDisconnectThreshold > 0 ? maxDisconnectThreshold : 1;
}
@Override
@@ -67,6 +80,15 @@ public class ZkStateChangeListener implements IZkStateListener
+ _zkHelixManager.getInstanceType());
_isConnected = false;
+ // Track the time stamp that the disconnected happens, then check history and see if
+ // we should disconnect the _zkHelixManager
+ _disconnectTimeHistory.add(System.currentTimeMillis());
+ if(isFlapping())
+ {
+ logger.error("isFlapping() returns true, so disconnect the helix manager. " + _zkHelixManager.getInstanceName() + " "
+ + _maxDisconnectThreshold + " disconnects in " + _timeWindowLengthMs + " Ms.");
+ _zkHelixManager.disconnectInternal();
+ }
break;
case Expired:
logger.info("KeeperState:" + keeperState + ", expiredSessionId: "
@@ -94,4 +116,23 @@ public class ZkStateChangeListener implements IZkStateListener
{
return _hasSessionExpired;
}
+
+ /**
+ * If zk state has changed into Disconnected for _maxDisconnectThreshold times during previous _timeWindowLengthMs Ms
+ * time window, we think that there are something wrong going on and disconnect the zkHelixManager from zk.
+ * */
+ boolean isFlapping()
+ {
+ if(_disconnectTimeHistory.size() == 0)
+ {
+ return false;
+ }
+ long mostRecentTimestamp = _disconnectTimeHistory.get(_disconnectTimeHistory.size() - 1);
+ // Remove disconnect history timestamp that are older than _timeWindowLengthMs ago
+ while((_disconnectTimeHistory.get(0) + _timeWindowLengthMs) < mostRecentTimestamp)
+ {
+ _disconnectTimeHistory.remove(0);
+ }
+ return _disconnectTimeHistory.size() > _maxDisconnectThreshold;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e690aff8/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
index 456505f..23f293b 100644
--- a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
@@ -66,7 +66,42 @@ public class ZkTestHelper
{
return _zkClient;
}
+ }
+
+ public static void disconnectSession(final ZkClient zkClient) throws Exception
+ {
+
+ ZkConnection connection = ((ZkConnection) zkClient.getConnection());
+ ZooKeeper curZookeeper = connection.getZookeeper();
+ LOG.info("Before expiry. sessionId: " + Long.toHexString(curZookeeper.getSessionId()));
+
+ Watcher watcher = new Watcher()
+ {
+ @Override
+ public void process(WatchedEvent event)
+ {
+ LOG.info("Process watchEvent: " + event);
+ }
+ };
+ final ZooKeeper dupZookeeper =
+ new ZooKeeper(connection.getServers(),
+ curZookeeper.getSessionTimeout(),
+ watcher,
+ curZookeeper.getSessionId(),
+ curZookeeper.getSessionPasswd());
+ // wait until connected, then close
+ while (dupZookeeper.getState() != States.CONNECTED)
+ {
+ Thread.sleep(10);
+ }
+ dupZookeeper.close();
+
+ connection = (ZkConnection) zkClient.getConnection();
+ curZookeeper = connection.getZookeeper();
+
+ // System.err.println("zk: " + oldZookeeper);
+ LOG.info("After expiry. sessionId: " + Long.toHexString(curZookeeper.getSessionId()));
}
public static void expireSession(final ZkClient zkClient) throws Exception
@@ -78,7 +113,7 @@ public class ZkTestHelper
@Override
public void handleStateChanged(KeeperState state) throws Exception
{
- // System.err.println("handleStateChanged. state: " + state);
+ System.err.println("handleStateChanged. state: " + state);
}
@Override
@@ -182,10 +217,8 @@ public class ZkTestHelper
+ expectState + ", op: " + op);
result = false;
}
-
}
}
-
}
}
return result;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e690aff8/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkDisconnectHistory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkDisconnectHistory.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkDisconnectHistory.java
new file mode 100644
index 0000000..6df52db
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkDisconnectHistory.java
@@ -0,0 +1,109 @@
+package org.apache.helix.manager.zk;
+
+import org.apache.helix.InstanceType;
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZkTestHelper;
+import org.apache.helix.ZkTestHelper.TestZkHelixManager;
+import org.apache.helix.integration.ZkIntegrationTestBase;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestZkDisconnectHistory extends ZkIntegrationTestBase
+{
+ @Test
+ public void testDisconnectHistory() throws Exception
+ {
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ final String clusterName = className + "_" + methodName;
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 10, // partitions per resource
+ 5, // number of nodes
+ 3, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+
+ String instanceName = "localhost_" + (12918 + 0);
+ TestZkHelixManager manager =
+ new TestZkHelixManager(clusterName,
+ instanceName,
+ InstanceType.PARTICIPANT,
+ ZK_ADDR);
+ manager.connect();
+ ZkClient zkClient = manager.getZkClient();
+ ZkTestHelper.expireSession(zkClient);
+ for(int i = 0;i < 4; i++)
+ {
+ ZkTestHelper.expireSession(zkClient);
+ Thread.sleep(500);
+ if(i < 5)
+ {
+ Assert.assertTrue(manager.isConnected());
+ }
+ }
+ ZkTestHelper.disconnectSession(zkClient);
+ for(int i = 0; i < 20; i++)
+ {
+ Thread.sleep(500);
+ if(!manager.isConnected()) break;
+ }
+ Assert.assertFalse(manager.isConnected());
+ }
+
+ @Test
+ public void testDisconnectFlappingWindow() throws Exception
+ {
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ final String clusterName = className + "_" + methodName;
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 10, // partitions per resource
+ 5, // number of nodes
+ 3, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+
+ // flapping time window to 5 sec
+ System.setProperty("helixmanager.flappingTimeWindow", "10000");
+ String instanceName = "localhost_" + (12918 + 1);
+ TestZkHelixManager manager2 =
+ new TestZkHelixManager(clusterName,
+ instanceName,
+ InstanceType.PARTICIPANT,
+ ZK_ADDR);
+ manager2.connect();
+ ZkClient zkClient = manager2.getZkClient();
+ for(int i = 0;i < 3; i++)
+ {
+ ZkTestHelper.expireSession(zkClient);
+ Thread.sleep(500);
+ Assert.assertTrue(manager2.isConnected());
+ }
+ Thread.sleep(10000);
+ // Old entries should be cleaned up
+ for(int i = 0;i < 5; i++)
+ {
+ ZkTestHelper.expireSession(zkClient);
+ Thread.sleep(500);
+ Assert.assertTrue(manager2.isConnected());
+ }
+ ZkTestHelper.disconnectSession(zkClient);
+ for(int i = 0; i < 20; i++)
+ {
+ Thread.sleep(500);
+ if(!manager2.isConnected()) break;
+ }
+ Assert.assertFalse(manager2.isConnected());
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e690aff8/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkStateChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkStateChangeListener.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkStateChangeListener.java
new file mode 100644
index 0000000..679614d
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkStateChangeListener.java
@@ -0,0 +1,80 @@
+package org.apache.helix.manager.zk;
+
+import org.apache.helix.TestHelper.StartCMResult;
+import org.apache.helix.integration.ZkStandAloneCMTestBaseWithPropertyServerCheck;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+
+public class TestZkStateChangeListener extends ZkStandAloneCMTestBaseWithPropertyServerCheck
+{
+ @Test
+ public void testDisconnectHistory() throws Exception
+ {
+ String controllerName = CONTROLLER_PREFIX + "_0";
+ StartCMResult controllerResult = _startCMResultMap.get(controllerName);
+ ZKHelixManager controller = (ZKHelixManager) controllerResult._manager;
+ ZkStateChangeListener listener1 = new ZkStateChangeListener(controller, 5000, 10);
+ // 11 disconnects in 5 sec
+ for(int i = 0;i < 11; i++)
+ {
+ Thread.sleep(200);
+ listener1.handleStateChanged(KeeperState.Disconnected);
+ if(i<10)
+ {
+ Assert.assertTrue(controller.isConnected());
+ }
+ else
+ {
+ Assert.assertFalse(controller.isConnected());
+ }
+ }
+
+ // If maxDisconnectThreshold is 0 it should be set to 1
+ String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + 0);
+ ZKHelixManager manager = (ZKHelixManager) _startCMResultMap.get(instanceName)._manager;
+
+ ZkStateChangeListener listener2 = new ZkStateChangeListener(manager, 5000, 0);
+ for(int i = 0;i < 2; i++)
+ {
+ Thread.sleep(200);
+ listener2.handleStateChanged(KeeperState.Disconnected);
+ if(i < 1)
+ {
+ Assert.assertTrue(manager.isConnected());
+ }
+ else
+ {
+ Assert.assertFalse(manager.isConnected());
+ }
+ }
+
+ // If there are long time after disconnect, older history should be cleanup
+ instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + 1);
+ manager = (ZKHelixManager) _startCMResultMap.get(instanceName)._manager;
+
+ ZkStateChangeListener listener3 = new ZkStateChangeListener(manager, 5000, 5);
+ for(int i = 0;i < 3; i++)
+ {
+ Thread.sleep(200);
+ listener3.handleStateChanged(KeeperState.Disconnected);
+ Assert.assertTrue(manager.isConnected());
+ }
+ Thread.sleep(5000);
+ // Old entries should be cleaned up
+ for(int i = 0;i < 3; i++)
+ {
+ Thread.sleep(200);
+ listener3.handleStateChanged(KeeperState.Disconnected);
+ Assert.assertTrue(manager.isConnected());
+ }
+ for(int i = 0;i < 2; i++)
+ {
+ Thread.sleep(200);
+ listener3.handleStateChanged(KeeperState.Disconnected);
+ Assert.assertTrue(manager.isConnected());
+ }
+ listener3.handleStateChanged(KeeperState.Disconnected);
+ Assert.assertFalse(manager.isConnected());
+ }
+}