You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ka...@apache.org on 2014/07/16 20:35:10 UTC
[1/3] git commit: [HELIX-471] Unregister ResourceMonitor on drop
Repository: helix
Updated Branches:
refs/heads/helix-0.6.x c37d1e81f -> 371e65769
[HELIX-471] Unregister ResourceMonitor on drop
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/a92a11d6
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/a92a11d6
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/a92a11d6
Branch: refs/heads/helix-0.6.x
Commit: a92a11d6a969bd848964cdf834ea3e3505269c7b
Parents: c37d1e8
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Fri Jul 11 16:12:34 2014 -0700
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Wed Jul 16 11:33:28 2014 -0700
----------------------------------------------------------------------
.../stages/ExternalViewComputeStage.java | 3 +
.../monitoring/mbeans/ClusterStatusMonitor.java | 18 ++++
.../mbeans/TestDropResourceMetricsReset.java | 106 +++++++++++++++++++
3 files changed, 127 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/a92a11d6/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
index ba1f2b5..903840c 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
@@ -116,6 +116,9 @@ public class ExternalViewComputeStage extends AbstractBaseStage {
clusterStatusMonitor.setResourceStatus(view,
cache._idealStateMap.get(view.getResourceName()), stateModelDef);
}
+ } else {
+ // Drop the metrics for the dropped resource
+ clusterStatusMonitor.unregisterResource(view.getResourceName());
}
// compare the new external view with current one, set only on different
http://git-wip-us.apache.org/repos/asf/helix/blob/a92a11d6/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java
index b4efec4..b4b5c01 100644
--- a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java
+++ b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/ClusterStatusMonitor.java
@@ -304,6 +304,24 @@ public class ClusterStatusMonitor implements ClusterStatusMonitorMBean {
}
}
+ /**
+ * Indicate that a resource has been dropped, thus making it OK to drop its metrics
+ * @param resourceName the resource that has been dropped
+ */
+ public void unregisterResource(String resourceName) {
+ if (_resourceMbeanMap.containsKey(resourceName)) {
+ synchronized (this) {
+ if (_resourceMbeanMap.containsKey(resourceName)) {
+ try {
+ unregisterResources(Arrays.asList(resourceName));
+ } catch (MalformedObjectNameException e) {
+ LOG.error("Could not unregister beans for " + resourceName, e);
+ }
+ }
+ }
+ }
+ }
+
public void setResourceStatus(ExternalView externalView, IdealState idealState,
StateModelDefinition stateModelDef) {
String topState = null;
http://git-wip-us.apache.org/repos/asf/helix/blob/a92a11d6/helix-core/src/test/java/org/apache/helix/monitoring/mbeans/TestDropResourceMetricsReset.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/monitoring/mbeans/TestDropResourceMetricsReset.java b/helix-core/src/test/java/org/apache/helix/monitoring/mbeans/TestDropResourceMetricsReset.java
new file mode 100644
index 0000000..850294d
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/monitoring/mbeans/TestDropResourceMetricsReset.java
@@ -0,0 +1,106 @@
+package org.apache.helix.monitoring.mbeans;
+
+/*
+ * 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.
+ */
+
+import java.lang.management.ManagementFactory;
+import java.util.Date;
+
+import javax.management.MBeanServerConnection;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.model.IdealState.RebalanceMode;
+import org.apache.helix.tools.ClusterSetup;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestDropResourceMetricsReset extends ZkUnitTestBase {
+ private static final MBeanServerConnection SERVER = ManagementFactory.getPlatformMBeanServer();
+
+ @Test
+ public void testBasic() throws Exception {
+ final int NUM_PARTICIPANTS = 4;
+ final int NUM_PARTITIONS = 64;
+ final int NUM_REPLICAS = 1;
+ final String RESOURCE_NAME = "TestDB0";
+
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+ System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));
+
+ // Set up cluster
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ NUM_PARTITIONS, // partitions per resource
+ NUM_PARTICIPANTS, // number of nodes
+ NUM_REPLICAS, // replicas
+ "MasterSlave", RebalanceMode.FULL_AUTO, // use FULL_AUTO mode to test node tagging
+ true); // do rebalance
+
+ // Start participants and controller
+ ClusterSetup setupTool = new ClusterSetup(_gZkClient);
+ MockParticipantManager[] participants = new MockParticipantManager[NUM_PARTICIPANTS];
+ for (int i = 0; i < NUM_PARTICIPANTS; i++) {
+ participants[i] =
+ new MockParticipantManager(ZK_ADDR, clusterName, "localhost_" + (12918 + i));
+ participants[i].syncStart();
+ }
+ ClusterControllerManager controller =
+ new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0");
+ controller.syncStart();
+
+ // Verify that the bean was created
+ Thread.sleep(1000);
+ ObjectName objectName = getObjectName(RESOURCE_NAME, clusterName);
+ Assert.assertTrue(SERVER.isRegistered(objectName));
+
+ // Drop the resource
+ setupTool.dropResourceFromCluster(clusterName, RESOURCE_NAME);
+
+ // Verify that the bean was removed
+ Thread.sleep(1000);
+ Assert.assertFalse(SERVER.isRegistered(objectName));
+
+ // Clean up
+ controller.syncStop();
+ for (MockParticipantManager participant : participants) {
+ participant.syncStop();
+ }
+ System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
+ }
+
+ private ObjectName getObjectName(String resourceName, String clusterName)
+ throws MalformedObjectNameException {
+ String clusterBeanName =
+ String.format("%s=%s", ClusterStatusMonitor.CLUSTER_DN_KEY, clusterName);
+ String resourceBeanName =
+ String.format("%s,%s=%s", clusterBeanName, ClusterStatusMonitor.RESOURCE_DN_KEY,
+ resourceName);
+ return new ObjectName(String.format("%s: %s", ClusterStatusMonitor.CLUSTER_STATUS_KEY,
+ resourceBeanName));
+ }
+}
[2/3] git commit: [HELIX-452] Increase frequency of status update
cleanup
Posted by ka...@apache.org.
[HELIX-452] Increase frequency of status update cleanup
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/abf4e496
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/abf4e496
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/abf4e496
Branch: refs/heads/helix-0.6.x
Commit: abf4e49637df67b24f9104d79f74940ee8869746
Parents: a92a11d
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Thu Jun 5 10:26:13 2014 -0700
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Wed Jul 16 11:34:06 2014 -0700
----------------------------------------------------------------------
.../controller/stages/BestPossibleStateCalcStage.java | 14 +++++++++-----
.../helix/manager/zk/ControllerManagerHelper.java | 2 ++
.../org/apache/helix/manager/zk/ZKHelixManager.java | 14 ++++++--------
3 files changed, 17 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/abf4e496/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
index 257a4b0..706fd41 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
@@ -63,11 +63,15 @@ public class BestPossibleStateCalcStage extends AbstractBaseStage {
compute(event, resourceMap, currentStateOutput);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
- ClusterStatusMonitor clusterStatusMonitor =
- (ClusterStatusMonitor) event.getAttribute("clusterStatusMonitor");
- if (clusterStatusMonitor != null) {
- clusterStatusMonitor.setPerInstanceResourceStatus(bestPossibleStateOutput,
- cache.getInstanceConfigMap(), resourceMap, cache.getStateModelDefMap());
+ try {
+ ClusterStatusMonitor clusterStatusMonitor =
+ (ClusterStatusMonitor) event.getAttribute("clusterStatusMonitor");
+ if (clusterStatusMonitor != null) {
+ clusterStatusMonitor.setPerInstanceResourceStatus(bestPossibleStateOutput,
+ cache.getInstanceConfigMap(), resourceMap, cache.getStateModelDefMap());
+ }
+ } catch (Exception e) {
+ logger.error("Could not update cluster status metrics!", e);
}
long endTime = System.currentTimeMillis();
http://git-wip-us.apache.org/repos/asf/helix/blob/abf4e496/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManagerHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManagerHelper.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManagerHelper.java
index d2b520b..9a817e3 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManagerHelper.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManagerHelper.java
@@ -56,10 +56,12 @@ public class ControllerManagerHelper {
new DefaultControllerMessageHandlerFactory();
_messagingService.getExecutor().registerMessageHandlerFactory(
defaultControllerMsgHandlerFactory.getMessageType(), defaultControllerMsgHandlerFactory);
+
MessageHandlerFactory defaultSchedulerMsgHandlerFactory =
new DefaultSchedulerMessageHandlerFactory(_manager);
_messagingService.getExecutor().registerMessageHandlerFactory(
defaultSchedulerMsgHandlerFactory.getMessageType(), defaultSchedulerMsgHandlerFactory);
+
MessageHandlerFactory defaultParticipantErrorMessageHandlerFactory =
new DefaultParticipantErrorMessageHandlerFactory(_manager);
_messagingService.getExecutor().registerMessageHandlerFactory(
http://git-wip-us.apache.org/repos/asf/helix/blob/abf4e496/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 01bbaf3..8d28dbd 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
@@ -36,11 +36,12 @@ import org.apache.helix.ControllerChangeListener;
import org.apache.helix.CurrentStateChangeListener;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerProperties;
import org.apache.helix.HelixTimerTask;
-import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.IdealStateChangeListener;
import org.apache.helix.InstanceConfigChangeListener;
import org.apache.helix.InstanceType;
@@ -48,7 +49,6 @@ import org.apache.helix.LiveInstanceChangeListener;
import org.apache.helix.LiveInstanceInfoProvider;
import org.apache.helix.MessageListener;
import org.apache.helix.PreConnectCallback;
-import org.apache.helix.HelixManagerProperties;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.PropertyPathConfig;
@@ -57,18 +57,16 @@ import org.apache.helix.ScopedConfigChangeListener;
import org.apache.helix.ZNRecord;
import org.apache.helix.controller.GenericHelixController;
import org.apache.helix.messaging.DefaultMessagingService;
-import org.apache.helix.model.ConfigScope;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.model.LiveInstance;
-import org.apache.helix.model.builder.ConfigScopeBuilder;
import org.apache.helix.monitoring.ZKPathDataDumpTask;
import org.apache.helix.participant.HelixStateMachineEngine;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.store.zk.AutoFallbackPropertyStore;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.log4j.Logger;
-import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper.States;
public class ZKHelixManager implements HelixManager, IZkStateListener {
@@ -139,9 +137,9 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
@Override
public void start() {
- long initialDelay = 30 * 60 * 1000;
- long period = 120 * 60 * 1000;
- int timeThresholdNoChange = 180 * 60 * 1000;
+ long initialDelay = 0;
+ long period = 15 * 60 * 1000;
+ int timeThresholdNoChange = 15 * 60 * 1000;
if (_timer == null) {
LOG.info("Start StatusDumpTask");
[3/3] git commit: [HELIX-472] Errors should be cleaned up less
frequently
Posted by ka...@apache.org.
[HELIX-472] Errors should be cleaned up less frequently
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/371e6576
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/371e6576
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/371e6576
Branch: refs/heads/helix-0.6.x
Commit: 371e65769774d4e116b97c0b88d52e2be545c8ec
Parents: abf4e49
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Tue Jul 15 11:33:43 2014 -0700
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Wed Jul 16 11:34:37 2014 -0700
----------------------------------------------------------------------
.../apache/helix/manager/zk/ZKHelixManager.java | 9 +-
.../helix/monitoring/ZKPathDataDumpTask.java | 31 +++---
.../helix/integration/TestSchedulerMessage.java | 2 +-
.../monitoring/TestZKPathDataDumpTask.java | 99 ++++++++++++++++++--
4 files changed, 119 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/371e6576/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 8d28dbd..f95f6ee 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
@@ -139,13 +139,16 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
public void start() {
long initialDelay = 0;
long period = 15 * 60 * 1000;
- int timeThresholdNoChange = 15 * 60 * 1000;
+ long timeThresholdNoChangeForStatusUpdates = 15 * 60 * 1000; // 15 minutes
+ long timeThresholdNoChangeForErrors = 24 * 60 * 60 * 1000; // 1 day
+ int maximumNumberOfLeafNodesAllowed = 10000;
if (_timer == null) {
LOG.info("Start StatusDumpTask");
_timer = new Timer("StatusDumpTimerTask", true);
- _timer.scheduleAtFixedRate(new ZKPathDataDumpTask(helixController, timeThresholdNoChange),
- initialDelay, period);
+ _timer.scheduleAtFixedRate(new ZKPathDataDumpTask(helixController,
+ timeThresholdNoChangeForStatusUpdates, timeThresholdNoChangeForErrors,
+ maximumNumberOfLeafNodesAllowed), initialDelay, period);
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/371e6576/helix-core/src/main/java/org/apache/helix/monitoring/ZKPathDataDumpTask.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/monitoring/ZKPathDataDumpTask.java b/helix-core/src/main/java/org/apache/helix/monitoring/ZKPathDataDumpTask.java
index a0190d2..0a91256 100644
--- a/helix-core/src/main/java/org/apache/helix/monitoring/ZKPathDataDumpTask.java
+++ b/helix-core/src/main/java/org/apache/helix/monitoring/ZKPathDataDumpTask.java
@@ -25,9 +25,9 @@ import java.util.TimerTask;
import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.PropertyType;
import org.apache.helix.ZNRecord;
-import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.util.HelixUtil;
import org.apache.log4j.Logger;
@@ -38,17 +38,24 @@ import com.google.common.collect.Lists;
public class ZKPathDataDumpTask extends TimerTask {
static Logger LOG = Logger.getLogger(ZKPathDataDumpTask.class);
- private final int _thresholdNoChangeInMs;
+ private final long _thresholdNoChangeMsForStatusUpdates;
+ private final long _thresholdNoChangeMsForErrors;
+ private final int _maxLeafCount;
private final HelixManager _manager;
private final ZNRecordSerializer _jsonSerializer;
- public ZKPathDataDumpTask(HelixManager manager, int thresholdNoChangeInMs) {
+ public ZKPathDataDumpTask(HelixManager manager, long thresholdNoChangeMsForStatusUpdates,
+ long thresholdNoChangeMsForErrors, int maxLeafCount) {
LOG.info("Init ZKPathDataDumpTask for cluster: " + manager.getClusterName()
- + ", thresholdNoChangeInMs: " + thresholdNoChangeInMs);
+ + ", thresholdNoChangeMsForStatusUpdates: " + thresholdNoChangeMsForStatusUpdates
+ + ", thresholdNoChangeMsForErrors: " + thresholdNoChangeMsForErrors + ", maxLeafCount: "
+ + maxLeafCount);
_manager = manager;
_jsonSerializer = new ZNRecordSerializer();
- _thresholdNoChangeInMs = thresholdNoChangeInMs;
+ _thresholdNoChangeMsForStatusUpdates = thresholdNoChangeMsForStatusUpdates;
+ _thresholdNoChangeMsForErrors = thresholdNoChangeMsForErrors;
+ _maxLeafCount = maxLeafCount;
}
@Override
@@ -70,25 +77,26 @@ public class ZKPathDataDumpTask extends TimerTask {
String statusUpdatePath =
HelixUtil.getInstancePropertyPath(_manager.getClusterName(), instance,
PropertyType.STATUSUPDATES);
- dump(baseAccessor, statusUpdatePath, _thresholdNoChangeInMs);
+ dump(baseAccessor, statusUpdatePath, _thresholdNoChangeMsForStatusUpdates, _maxLeafCount);
// dump participant errors
String errorPath =
HelixUtil.getInstancePropertyPath(_manager.getClusterName(), instance,
PropertyType.ERRORS);
- dump(baseAccessor, errorPath, _thresholdNoChangeInMs * 3);
+ dump(baseAccessor, errorPath, _thresholdNoChangeMsForErrors, _maxLeafCount);
}
// dump controller status updates
String controllerStatusUpdatePath =
HelixUtil.getControllerPropertyPath(_manager.getClusterName(),
PropertyType.STATUSUPDATES_CONTROLLER);
- dump(baseAccessor, controllerStatusUpdatePath, _thresholdNoChangeInMs);
+ dump(baseAccessor, controllerStatusUpdatePath, _thresholdNoChangeMsForStatusUpdates,
+ _maxLeafCount);
// dump controller errors
String controllerErrorPath =
HelixUtil.getControllerPropertyPath(_manager.getClusterName(),
PropertyType.ERRORS_CONTROLLER);
- dump(baseAccessor, controllerErrorPath, _thresholdNoChangeInMs);
+ dump(baseAccessor, controllerErrorPath, _thresholdNoChangeMsForErrors, _maxLeafCount);
}
/**
@@ -122,7 +130,8 @@ public class ZKPathDataDumpTask extends TimerTask {
return leafPaths;
}
- void dump(BaseDataAccessor<ZNRecord> accessor, String ancestorPath, int threshold) {
+ void dump(BaseDataAccessor<ZNRecord> accessor, String ancestorPath, long threshold,
+ int maxLeafCount) {
List<String> leafPaths = scanPath(accessor, ancestorPath);
if (leafPaths.isEmpty()) {
return;
@@ -133,7 +142,7 @@ public class ZKPathDataDumpTask extends TimerTask {
long now = System.currentTimeMillis();
for (int i = 0; i < stats.length; i++) {
Stat stat = stats[i];
- if ((now - stat.getMtime()) > threshold) {
+ if ((stats.length > maxLeafCount) || ((now - stat.getMtime()) > threshold)) {
dumpPaths.add(leafPaths.get(i));
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/371e6576/helix-core/src/test/java/org/apache/helix/integration/TestSchedulerMessage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestSchedulerMessage.java b/helix-core/src/test/java/org/apache/helix/integration/TestSchedulerMessage.java
index cb6e186..70713f3 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestSchedulerMessage.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestSchedulerMessage.java
@@ -393,7 +393,7 @@ public class TestSchedulerMessage extends ZkStandAloneCMTestBase {
}
}
Thread.sleep(3000);
- ZKPathDataDumpTask dumpTask = new ZKPathDataDumpTask(manager, 0);
+ ZKPathDataDumpTask dumpTask = new ZKPathDataDumpTask(manager, 0L, 0L, Integer.MAX_VALUE);
dumpTask.run();
subPaths = _gZkClient.getChildren(controllerStatusPath);
http://git-wip-us.apache.org/repos/asf/helix/blob/371e6576/helix-core/src/test/java/org/apache/helix/monitoring/TestZKPathDataDumpTask.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/monitoring/TestZKPathDataDumpTask.java b/helix-core/src/test/java/org/apache/helix/monitoring/TestZKPathDataDumpTask.java
index a3d8ae3..d073dd2 100644
--- a/helix-core/src/test/java/org/apache/helix/monitoring/TestZKPathDataDumpTask.java
+++ b/helix-core/src/test/java/org/apache/helix/monitoring/TestZKPathDataDumpTask.java
@@ -19,6 +19,9 @@ package org.apache.helix.monitoring;
* under the License.
*/
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import java.util.Date;
import org.apache.helix.BaseDataAccessor;
@@ -30,14 +33,11 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.ZkUnitTestBase;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
-import org.apache.helix.model.StatusUpdate;
import org.apache.helix.model.Error;
+import org.apache.helix.model.StatusUpdate;
import org.testng.Assert;
import org.testng.annotations.Test;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
public class TestZKPathDataDumpTask extends ZkUnitTestBase {
@Test
@@ -67,8 +67,9 @@ public class TestZKPathDataDumpTask extends ZkUnitTestBase {
when(manager.getHelixDataAccessor()).thenReturn(accessor);
when(manager.getClusterName()).thenReturn(clusterName);
- // run dump task without statusUpdates and errors, should not remove any existing statusUpdate/error paths
- ZKPathDataDumpTask task = new ZKPathDataDumpTask(manager, 0);
+ // run dump task without statusUpdates and errors, should not remove any existing
+ // statusUpdate/error paths
+ ZKPathDataDumpTask task = new ZKPathDataDumpTask(manager, 0L, 0L, Integer.MAX_VALUE);
task.run();
PropertyKey controllerStatusUpdateKey = keyBuilder.controllerTaskStatuses();
Assert.assertTrue(baseAccessor.exists(controllerStatusUpdateKey.getPath(), 0));
@@ -88,7 +89,8 @@ public class TestZKPathDataDumpTask extends ZkUnitTestBase {
// add controller status updates and errors
controllerStatusUpdateKey = keyBuilder.controllerTaskStatus("session_0", "TestDB");
- accessor.setProperty(controllerStatusUpdateKey, new StatusUpdate(new ZNRecord("controllerStatusUpdate")));
+ accessor.setProperty(controllerStatusUpdateKey, new StatusUpdate(new ZNRecord(
+ "controllerStatusUpdate")));
controllerErrorKey = keyBuilder.controllerTaskError("TestDB_error");
accessor.setProperty(controllerErrorKey, new Error(new ZNRecord("controllerError")));
@@ -110,4 +112,87 @@ public class TestZKPathDataDumpTask extends ZkUnitTestBase {
System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
}
+
+ @Test
+ public void testCapacityReached() throws Exception {
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+ int n = 1;
+
+ System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 2, // partitions per resource
+ n, // number of nodes
+ 1, // replicas
+ "MasterSlave", true); // do rebalance
+
+ HelixDataAccessor accessor =
+ new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ BaseDataAccessor<ZNRecord> baseAccessor = accessor.getBaseDataAccessor();
+
+ HelixManager manager = mock(HelixManager.class);
+ when(manager.getHelixDataAccessor()).thenReturn(accessor);
+ when(manager.getClusterName()).thenReturn(clusterName);
+
+ // run dump task without statusUpdates and errors, should not remove any existing
+ // statusUpdate/error paths
+ ZKPathDataDumpTask task = new ZKPathDataDumpTask(manager, Long.MAX_VALUE, Long.MAX_VALUE, 1);
+ task.run();
+ PropertyKey controllerStatusUpdateKey = keyBuilder.controllerTaskStatuses();
+ Assert.assertTrue(baseAccessor.exists(controllerStatusUpdateKey.getPath(), 0));
+ PropertyKey controllerErrorKey = keyBuilder.controllerTaskErrors();
+ Assert.assertTrue(baseAccessor.exists(controllerErrorKey.getPath(), 0));
+ PropertyKey statusUpdateKey = keyBuilder.stateTransitionStatus("localhost_12918");
+ Assert.assertTrue(baseAccessor.exists(statusUpdateKey.getPath(), 0));
+ PropertyKey errorKey = keyBuilder.stateTransitionErrors("localhost_12918");
+ Assert.assertTrue(baseAccessor.exists(errorKey.getPath(), 0));
+
+ // add participant status updates and errors
+ statusUpdateKey =
+ keyBuilder.stateTransitionStatus("localhost_12918", "session_0", "TestDB0", "TestDB0_0");
+ accessor.setProperty(statusUpdateKey, new StatusUpdate(new ZNRecord("statusUpdate")));
+ errorKey =
+ keyBuilder.stateTransitionError("localhost_12918", "session_0", "TestDB0", "TestDB0_0");
+ accessor.setProperty(errorKey, new Error(new ZNRecord("error")));
+
+ // add controller status updates and errors (one of each, should not trigger anything)
+ controllerStatusUpdateKey = keyBuilder.controllerTaskStatus("session_0", "TestDB");
+ accessor.setProperty(controllerStatusUpdateKey, new StatusUpdate(new ZNRecord(
+ "controllerStatusUpdate")));
+ controllerErrorKey = keyBuilder.controllerTaskError("TestDB_error");
+ accessor.setProperty(controllerErrorKey, new Error(new ZNRecord("controllerError")));
+
+ // run dump task, should not remove anything because the threshold is not exceeded
+ task.run();
+ Assert.assertTrue(baseAccessor.exists(controllerStatusUpdateKey.getPath(), 0));
+ Assert.assertTrue(baseAccessor.exists(controllerErrorKey.getPath(), 0));
+ Assert.assertTrue(baseAccessor.exists(statusUpdateKey.getPath(), 0));
+ Assert.assertTrue(baseAccessor.exists(errorKey.getPath(), 0));
+
+ // add a second set of all status updates and errors
+ statusUpdateKey =
+ keyBuilder.stateTransitionStatus("localhost_12918", "session_0", "TestDB0", "TestDB0_1");
+ accessor.setProperty(statusUpdateKey, new StatusUpdate(new ZNRecord("statusUpdate")));
+ errorKey =
+ keyBuilder.stateTransitionError("localhost_12918", "session_0", "TestDB0", "TestDB0_1");
+ accessor.setProperty(errorKey, new Error(new ZNRecord("error")));
+ controllerStatusUpdateKey = keyBuilder.controllerTaskStatus("session_0", "TestDB1");
+ accessor.setProperty(controllerStatusUpdateKey, new StatusUpdate(new ZNRecord(
+ "controllerStatusUpdate")));
+ controllerErrorKey = keyBuilder.controllerTaskError("TestDB1_error");
+ accessor.setProperty(controllerErrorKey, new Error(new ZNRecord("controllerError")));
+
+ // run dump task, should remove everything since capacities are exceeded
+ task.run();
+ Assert.assertFalse(baseAccessor.exists(controllerStatusUpdateKey.getPath(), 0));
+ Assert.assertFalse(baseAccessor.exists(controllerErrorKey.getPath(), 0));
+ Assert.assertFalse(baseAccessor.exists(statusUpdateKey.getPath(), 0));
+ Assert.assertFalse(baseAccessor.exists(errorKey.getPath(), 0));
+ }
}