You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2019/02/26 19:45:47 UTC
[helix] branch master updated (150d12f -> 05a1286)
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git.
from 150d12f [HELIX-805] Implementation on HelixAdmin to check if cluster in maintenance mode
new b0f6150 Add ERROR mapping for displaying in BestPossible and Intemediate assignment.
new abd91cc Revert back for N -> N+1 fix
new 05a1286 Ignore register mbean request if the MBean has been registered.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rebalancer/DelayedAutoRebalancer.java | 13 +-
.../helix/monitoring/mbeans/MBeanRegistrar.java | 2 +
.../mbeans/dynamicMBeans/DynamicMBeanProvider.java | 43 ++++--
.../java/org/apache/helix/common/ZkTestBase.java | 9 ++
.../helix/integration/TestErrorReplicaPersist.java | 151 +++++++++++++++++++++
.../TestDelayedAutoRebalancer.OnlineOffline.json | 2 +-
6 files changed, 204 insertions(+), 16 deletions(-)
create mode 100644 helix-core/src/test/java/org/apache/helix/integration/TestErrorReplicaPersist.java
[helix] 01/03: Add ERROR mapping for displaying in BestPossible and
Intemediate assignment.
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit b0f61507915f5052f53854ddbefcacac86f9c8d4
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Fri Feb 1 13:23:35 2019 -0800
Add ERROR mapping for displaying in BestPossible and Intemediate assignment.
With adding this ERROR mapping replica, it should not break any rules of assignment. It was purely serving for showing the ERROR partition targeted host.
---
.../rebalancer/DelayedAutoRebalancer.java | 21 ++-
.../java/org/apache/helix/common/ZkTestBase.java | 9 ++
.../helix/integration/TestErrorReplicaPersist.java | 151 +++++++++++++++++++++
.../TestDelayedAutoRebalancer.OnlineOffline.json | 2 +-
4 files changed, 178 insertions(+), 5 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
index 7dd9128..62ad37f 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
@@ -529,24 +529,37 @@ public class DelayedAutoRebalancer extends AbstractRebalancer<ResourceController
// we should drop all partitions from previous assigned instances.
if (!currentMapWithPreferenceList.values().contains(HelixDefinedState.ERROR.name())
&& bestPossibleStateMap.size() > numReplicas && readyToDrop(currentStateMap,
- bestPossibleStateMap, numReplicas, combinedPreferenceList)) {
+ bestPossibleStateMap, preferenceList, combinedPreferenceList)) {
for (int i = 0; i < combinedPreferenceList.size() - numReplicas; i++) {
String instanceToDrop = combinedPreferenceList.get(combinedPreferenceList.size() - i - 1);
bestPossibleStateMap.put(instanceToDrop, HelixDefinedState.DROPPED.name());
}
}
+ // Adding ERROR replica mapping to best possible
+ // ERROR assignment should be mutual excluded from DROPPED assignment because
+ // once there is an ERROR replica in the mapping, bestPossibleStateMap.size() > numReplicas prevents
+ // code entering the DROPPING stage.
+ for (String instance : combinedPreferenceList) {
+ if (currentStateMap.containsKey(instance) && currentStateMap.get(instance)
+ .equals(HelixDefinedState.ERROR.name())) {
+ bestPossibleStateMap.put(instance, HelixDefinedState.ERROR.name());
+ }
+ }
+
return bestPossibleStateMap;
}
private boolean readyToDrop(Map<String, String> currentStateMap,
- Map<String, String> bestPossibleMap, int numReplicas, List<String> combinedPreferenceList) {
+ Map<String, String> bestPossibleMap, List<String> preferenceList,
+ List<String> combinedPreferenceList) {
if (currentStateMap.size() != bestPossibleMap.size()) {
return false;
}
+ Set<String> tmpPreferenceSet = new HashSet<>(preferenceList);
+ tmpPreferenceSet.retainAll(combinedPreferenceList);
- for (int i = 0; i < numReplicas; i++) {
- String instance = combinedPreferenceList.get(i);
+ for (String instance : tmpPreferenceSet) {
if (!currentStateMap.containsKey(instance) || !currentStateMap.get(instance)
.equals(bestPossibleMap.get(instance))) {
return false;
diff --git a/helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java b/helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java
index 700d204..721cd33 100644
--- a/helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java
+++ b/helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java
@@ -244,6 +244,15 @@ public class ZkTestBase {
configAccessor.setInstanceConfig(clusterName, instanceName, instanceConfig);
}
+ protected void enableDelayRebalanceInCluster(HelixZkClient zkClient, String clusterName,
+ boolean enabled, long delay) {
+ ConfigAccessor configAccessor = new ConfigAccessor(zkClient);
+ ClusterConfig clusterConfig = configAccessor.getClusterConfig(clusterName);
+ clusterConfig.setDelayRebalaceEnabled(enabled);
+ clusterConfig.setRebalanceDelayTime(delay);
+ configAccessor.setClusterConfig(clusterName, clusterConfig);
+ }
+
protected void enableP2PInCluster(String clusterName, ConfigAccessor configAccessor,
boolean enable) {
// enable p2p message in cluster.
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestErrorReplicaPersist.java b/helix-core/src/test/java/org/apache/helix/integration/TestErrorReplicaPersist.java
new file mode 100644
index 0000000..126175a
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestErrorReplicaPersist.java
@@ -0,0 +1,151 @@
+package org.apache.helix.integration;
+
+import java.util.Date;
+import org.apache.helix.HelixRollbackException;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.controller.rebalancer.DelayedAutoRebalancer;
+import org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy;
+import org.apache.helix.integration.common.ZkStandAloneCMTestBase;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.integration.rebalancer.TestAutoRebalance;
+import org.apache.helix.mock.participant.MockDelayMSStateModel;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.MasterSlaveSMD;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.StateMachineEngine;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier;
+import org.apache.helix.tools.ClusterVerifiers.HelixClusterVerifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class TestErrorReplicaPersist extends ZkStandAloneCMTestBase {
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ // Logger.getRootLogger().setLevel(Level.INFO);
+ System.out.println("START " + CLASS_NAME + " at " + new Date(System.currentTimeMillis()));
+
+ int numNode = NODE_NR + 1;
+ _participants = new MockParticipantManager[numNode];
+ // setup storage cluster
+ _gSetupTool.addCluster(CLUSTER_NAME, true);
+ createResourceWithDelayedRebalance(CLUSTER_NAME, TEST_DB, MasterSlaveSMD.name, _PARTITIONS,
+ _replica, _replica - 1, 1800000, CrushEdRebalanceStrategy.class.getName());
+ for (int i = 0; i < numNode; i++) {
+ String storageNodeName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
+ _gSetupTool.addInstanceToCluster(CLUSTER_NAME, storageNodeName);
+ }
+
+ _gSetupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, _replica);
+
+ // start dummy participants
+ for (int i = 0; i < numNode; i++) {
+ String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
+ MockParticipantManager participant =
+ new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+ participant.syncStart();
+ _participants[i] = participant;
+ }
+ enablePersistBestPossibleAssignment(_gZkClient, CLUSTER_NAME, true);
+ enableDelayRebalanceInCluster(_gZkClient, CLUSTER_NAME, true, 1800000);
+
+ // start controller
+ String controllerName = CONTROLLER_PREFIX + "_0";
+ _controller = new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, controllerName);
+ _controller.syncStart();
+
+ boolean result = ClusterStateVerifier.verifyByZkCallback(
+ new TestAutoRebalance.ExternalViewBalancedVerifier(_gZkClient, CLUSTER_NAME, TEST_DB));
+
+ Assert.assertTrue(result);
+ }
+
+ @AfterClass
+ public void afterClass() throws Exception {
+ for (MockParticipantManager participant : _participants) {
+ participant.syncStop();
+ }
+ super.afterClass();
+ }
+
+ @Test
+ public void testErrorReplicaPersist() throws InterruptedException {
+ for (int i = 0; i < (NODE_NR + 1) / 2; i++) {
+ _participants[i].syncStop();
+ Thread.sleep(2000);
+ String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
+ MockParticipantManager participant =
+ new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+ StateMachineEngine stateMachineEngine = participant.getStateMachineEngine();
+ stateMachineEngine
+ .registerStateModelFactory(MasterSlaveSMD.name, new MockFailedMSStateModelFactory());
+ participant.syncStart();
+ _participants[i] = participant;
+ }
+ HelixClusterVerifier verifier =
+ new BestPossibleExternalViewVerifier.Builder(CLUSTER_NAME).setZkClient(_gZkClient).build();
+ Assert.assertTrue(((BestPossibleExternalViewVerifier) verifier).verifyByPolling());
+ for (int i = 0; i < (NODE_NR + 1) / 2; i++) {
+ _gSetupTool.getClusterManagementTool()
+ .enableInstance(CLUSTER_NAME, _participants[i].getInstanceName(), false);
+ }
+
+ Assert.assertTrue(((BestPossibleExternalViewVerifier) verifier).verifyByPolling());
+ }
+
+
+ class MockFailedMSStateModelFactory
+ extends StateModelFactory<MockFailedMSStateModel> {
+
+ @Override
+ public MockFailedMSStateModel createNewStateModel(String resourceName,
+ String partitionKey) {
+ MockFailedMSStateModel model = new MockFailedMSStateModel();
+ return model;
+ }
+ }
+
+ @StateModelInfo(initialState = "OFFLINE", states = { "MASTER", "SLAVE", "ERROR"
+ })
+ public static class MockFailedMSStateModel extends StateModel {
+ private static Logger LOG = LoggerFactory.getLogger(MockFailedMSStateModel.class);
+
+ public MockFailedMSStateModel() {
+ }
+
+ @Transition(to = "SLAVE", from = "OFFLINE") public void onBecomeSlaveFromOffline(
+ Message message, NotificationContext context) throws IllegalAccessException {
+ throw new IllegalAccessException("Failed!");
+ }
+
+ @Transition(to = "MASTER", from = "SLAVE") public void onBecomeMasterFromSlave(Message message,
+ NotificationContext context) throws InterruptedException, HelixRollbackException {
+ LOG.error("Become MASTER from SLAVE");
+ }
+
+ @Transition(to = "SLAVE", from = "MASTER") public void onBecomeSlaveFromMaster(Message message,
+ NotificationContext context) {
+ LOG.info("Become Slave from Master");
+ }
+
+ @Transition(to = "OFFLINE", from = "SLAVE") public void onBecomeOfflineFromSlave(
+ Message message, NotificationContext context) {
+ LOG.info("Become OFFLINE from SLAVE");
+ }
+
+ @Transition(to = "DROPPED", from = "OFFLINE") public void onBecomeDroppedFromOffline(
+ Message message, NotificationContext context) {
+ LOG.info("Become DROPPED FROM OFFLINE");
+ }
+ }
+
+}
diff --git a/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json b/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
index 9ae6b24..55f5b3c 100644
--- a/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
+++ b/helix-core/src/test/resources/TestDelayedAutoRebalancer.OnlineOffline.json
@@ -53,7 +53,7 @@
"bestPossibleStates": {
"localhost_2": "ONLINE",
"localhost_3": "ONLINE",
- "localhost_0": "DROPPED",
+ "localhost_0": "ERROR",
"localhost_1": "ONLINE"
}
},
[helix] 03/03: Ignore register mbean request if the MBean has been
registered.
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 05a1286bdb48ccf7f8d0c6d80f783c8f85379839
Author: Jiajun Wang <jj...@linkedin.com>
AuthorDate: Wed Feb 6 18:02:33 2019 -0800
Ignore register mbean request if the MBean has been registered.
This is mainly for reducing unnessary logs.
The additional change is to support more flexible MBean provider usage. We used to restrict a provider to be single use. If multiple registers happen, a warning message will be printed. This logic is no longer necessary with the previous DynamicMBeanProvider simplification and improvements.
---
.../helix/monitoring/mbeans/MBeanRegistrar.java | 2 +
.../mbeans/dynamicMBeans/DynamicMBeanProvider.java | 43 +++++++++++++++-------
2 files changed, 31 insertions(+), 14 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/MBeanRegistrar.java b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/MBeanRegistrar.java
index 31b7388..00db47f 100644
--- a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/MBeanRegistrar.java
+++ b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/MBeanRegistrar.java
@@ -52,6 +52,7 @@ public class MBeanRegistrar {
num++;
try {
_beanServer.registerMBean(object, newObjectName);
+ LOG.info("MBean {} has been registered.", newObjectName.getCanonicalName());
} catch (InstanceAlreadyExistsException e) {
continue;
} catch (JMException e) {
@@ -81,6 +82,7 @@ public class MBeanRegistrar {
if (objectName != null && _beanServer.isRegistered(objectName)) {
try {
_beanServer.unregisterMBean(objectName);
+ LOG.info("MBean {} has been un-registered.", objectName.getCanonicalName());
} catch (JMException e) {
LOG.warn("Error in un-registering: " + objectName.getCanonicalName(), e);
}
diff --git a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/dynamicMBeans/DynamicMBeanProvider.java b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/dynamicMBeans/DynamicMBeanProvider.java
index fbbb9e6..4299159 100644
--- a/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/dynamicMBeans/DynamicMBeanProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/monitoring/mbeans/dynamicMBeans/DynamicMBeanProvider.java
@@ -19,15 +19,32 @@ package org.apache.helix.monitoring.mbeans.dynamicMBeans;
* under the License.
*/
-import org.apache.helix.HelixException;
+import javax.management.Attribute;
+import javax.management.AttributeList;
+import javax.management.AttributeNotFoundException;
+import javax.management.DynamicMBean;
+import javax.management.InvalidAttributeValueException;
+import javax.management.JMException;
+import javax.management.MBeanAttributeInfo;
+import javax.management.MBeanConstructorInfo;
+import javax.management.MBeanException;
+import javax.management.MBeanInfo;
+import javax.management.MBeanNotificationInfo;
+import javax.management.MBeanOperationInfo;
+import javax.management.ObjectName;
+import javax.management.ReflectionException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
import org.apache.helix.monitoring.SensorNameProvider;
import org.apache.helix.monitoring.mbeans.MBeanRegistrar;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.management.*;
-import java.util.*;
-
/**
* Dynamic MBean provider that reporting DynamicMetric attributes
*/
@@ -40,7 +57,7 @@ public abstract class DynamicMBeanProvider implements DynamicMBean, SensorNamePr
// Attribute name to the DynamicMetric object mapping
private final Map<String, DynamicMetric> _attributeMap = new HashMap<>();
- private ObjectName _objectName;
+ private ObjectName _objectName = null;
private MBeanInfo _mBeanInfo;
/**
@@ -53,13 +70,8 @@ public abstract class DynamicMBeanProvider implements DynamicMBean, SensorNamePr
*/
protected synchronized boolean doRegister(Collection<DynamicMetric<?, ?>> dynamicMetrics,
String description, String domain, String... keyValuePairs) throws JMException {
- if (_objectName != null) {
- _logger.warn("Mbean has been registered before. Please create new object for new registration.");
- return false;
- }
- updateAttributtInfos(dynamicMetrics, description);
- _objectName = MBeanRegistrar.register(this, domain, keyValuePairs);
- return true;
+ return doRegister(dynamicMetrics, description,
+ MBeanRegistrar.buildObjectName(domain, keyValuePairs));
}
/**
@@ -72,7 +84,8 @@ public abstract class DynamicMBeanProvider implements DynamicMBean, SensorNamePr
protected synchronized boolean doRegister(Collection<DynamicMetric<?, ?>> dynamicMetrics,
String description, ObjectName objectName) throws JMException {
if (_objectName != null) {
- _logger.warn("Mbean has been registered before. Please create new object for new registration.");
+ _logger.debug("Mbean {} has already been registered. Ignore register request.",
+ objectName.getCanonicalName());
return false;
}
updateAttributtInfos(dynamicMetrics, description);
@@ -136,10 +149,12 @@ public abstract class DynamicMBeanProvider implements DynamicMBean, SensorNamePr
public abstract DynamicMBeanProvider register() throws JMException;
/**
- * After unregistered, the MBean can't be registered again, a new monitor has to be created.
+ * Unregister the MBean and clean up object name record.
+ * Note that all the metric data is kept even after unregister.
*/
public synchronized void unregister() {
MBeanRegistrar.unregister(_objectName);
+ _objectName = null;
}
@Override
[helix] 02/03: Revert back for N -> N+1 fix
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit abd91cc3757d6bcb99f07b288264d68dbbf8cdb4
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Wed Feb 6 12:08:46 2019 -0800
Revert back for N -> N+1 fix
---
.../helix/controller/rebalancer/DelayedAutoRebalancer.java | 12 +++++-------
1 file changed, 5 insertions(+), 7 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
index 62ad37f..1d00d60 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/DelayedAutoRebalancer.java
@@ -529,7 +529,7 @@ public class DelayedAutoRebalancer extends AbstractRebalancer<ResourceController
// we should drop all partitions from previous assigned instances.
if (!currentMapWithPreferenceList.values().contains(HelixDefinedState.ERROR.name())
&& bestPossibleStateMap.size() > numReplicas && readyToDrop(currentStateMap,
- bestPossibleStateMap, preferenceList, combinedPreferenceList)) {
+ bestPossibleStateMap, numReplicas, combinedPreferenceList)) {
for (int i = 0; i < combinedPreferenceList.size() - numReplicas; i++) {
String instanceToDrop = combinedPreferenceList.get(combinedPreferenceList.size() - i - 1);
bestPossibleStateMap.put(instanceToDrop, HelixDefinedState.DROPPED.name());
@@ -551,15 +551,13 @@ public class DelayedAutoRebalancer extends AbstractRebalancer<ResourceController
}
private boolean readyToDrop(Map<String, String> currentStateMap,
- Map<String, String> bestPossibleMap, List<String> preferenceList,
- List<String> combinedPreferenceList) {
+ Map<String, String> bestPossibleMap, int numReplicas, List<String> combinedPreferenceList) {
if (currentStateMap.size() != bestPossibleMap.size()) {
return false;
}
- Set<String> tmpPreferenceSet = new HashSet<>(preferenceList);
- tmpPreferenceSet.retainAll(combinedPreferenceList);
-
- for (String instance : tmpPreferenceSet) {
+
+ for (int i = 0; i < numReplicas; i++) {
+ String instance = combinedPreferenceList.get(i);
if (!currentStateMap.containsKey(instance) || !currentStateMap.get(instance)
.equals(bestPossibleMap.get(instance))) {
return false;