You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ji...@apache.org on 2020/03/11 17:55:54 UTC
[helix] branch master updated: Migrate the IdealState usage to read
Resource Config for the delayed rebalance. (#878)
This is an automated email from the ASF dual-hosted git repository.
jiajunwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new e5c2a23 Migrate the IdealState usage to read Resource Config for the delayed rebalance. (#878)
e5c2a23 is described below
commit e5c2a2332231c03c87b1f03f6e3b4715c8910c7f
Author: Jiajun Wang <18...@users.noreply.github.com>
AuthorDate: Wed Mar 11 10:55:47 2020 -0700
Migrate the IdealState usage to read Resource Config for the delayed rebalance. (#878)
* Currently, the same configuration item can be configured in both Resource Config and Ideal State. In theory, the Resource Config is the right place.
This is the first step to migrate the IdealState usage to read the Resource Config.
Moving forward, IdealState should not be a method for the controller to take input. And any ideal state update to the IS nodes won't trigger a rebalance pipeline.
* Add setXXXIfAbsent methods to the ZNRecord for simplifying code.
* Add redundant parameter IdealState to the DelayedRebalanceUtil.getMinActiveReplica() before we fully migrate the configs to the resource config to avoid incorrect usage.
---
.../rebalancer/DelayedAutoRebalancer.java | 7 +-
.../rebalancer/util/DelayedRebalanceUtil.java | 23 +++++-
.../rebalancer/waged/WagedRebalancer.java | 9 +-
.../waged/model/ClusterModelProvider.java | 33 +-------
.../org/apache/helix/model/ResourceConfig.java | 70 +++++++++++++++-
.../org/apache/helix/model/TestResourceConfig.java | 95 ++++++++++++++++++++--
.../apache/helix/zookeeper/datamodel/ZNRecord.java | 30 +++++++
7 files changed, 220 insertions(+), 47 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 448cc73..1997cfa 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
@@ -30,7 +30,6 @@ import java.util.Map;
import java.util.Set;
import org.apache.helix.HelixDefinedState;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.api.config.StateTransitionThrottleConfig;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.rebalancer.util.DelayedRebalanceUtil;
@@ -42,6 +41,7 @@ import org.apache.helix.model.Resource;
import org.apache.helix.model.ResourceAssignment;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -167,8 +167,9 @@ public class DelayedAutoRebalancer extends AbstractRebalancer<ResourceController
if (DelayedRebalanceUtil.isDelayRebalanceEnabled(currentIdealState, clusterConfig)) {
List<String> activeNodeList = new ArrayList<>(activeNodes);
Collections.sort(activeNodeList);
- int minActiveReplicas =
- DelayedRebalanceUtil.getMinActiveReplica(currentIdealState, replicaCount);
+ int minActiveReplicas = DelayedRebalanceUtil.getMinActiveReplica(
+ ResourceConfig.mergeIdealStateWithResourceConfig(resourceConfig, currentIdealState),
+ currentIdealState, replicaCount);
ZNRecord newActiveMapping = _rebalanceStrategy
.computePartitionAssignment(allNodeList, activeNodeList, currentMapping, clusterData);
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/DelayedRebalanceUtil.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/DelayedRebalanceUtil.java
index 1342860..0bb6d59 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/DelayedRebalanceUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/DelayedRebalanceUtil.java
@@ -25,10 +25,12 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+
import org.apache.helix.HelixManager;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.ResourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -209,13 +211,26 @@ public class DelayedRebalanceUtil {
/**
* Get the minimum active replica count threshold that allows delayed rebalance.
+ * Prioritize of the input params:
+ * 1. resourceConfig
+ * 2. idealState
+ * 3. replicaCount
+ * The lower priority minimum active replica count will only be applied if the higher priority
+ * items are missing.
+ * TODO: Remove the idealState input once we have all the config information migrated to the
+ * TODO: resource config by default.
*
- * @param idealState the resource Ideal State
- * @param replicaCount the expected active replica count.
+ * @param resourceConfig the resource config
+ * @param idealState the ideal state of the resource
+ * @param replicaCount the expected active replica count.
* @return the expected minimum active replica count that is required
*/
- public static int getMinActiveReplica(IdealState idealState, int replicaCount) {
- int minActiveReplicas = idealState.getMinActiveReplicas();
+ public static int getMinActiveReplica(ResourceConfig resourceConfig, IdealState idealState,
+ int replicaCount) {
+ int minActiveReplicas = resourceConfig == null ? -1 : resourceConfig.getMinActiveReplica();
+ if (minActiveReplicas < 0) {
+ minActiveReplicas = idealState.getMinActiveReplicas();
+ }
if (minActiveReplicas < 0) {
minActiveReplicas = replicaCount;
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
index 8a21bbb..86c4d09 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java
@@ -746,10 +746,11 @@ public class WagedRebalancer implements StatefulRebalancer<ResourceControllerDat
IdealState currentIdealState = clusterData.getIdealState(resourceName);
Set<String> enabledLiveInstances = clusterData.getEnabledLiveInstances();
int numReplica = currentIdealState.getReplicaCount(enabledLiveInstances.size());
- int minActiveReplica =
- DelayedRebalanceUtil.getMinActiveReplica(currentIdealState, numReplica);
- Map<String, List<String>> finalPreferenceLists = DelayedRebalanceUtil
- .getFinalDelayedMapping(newActiveIdealState.getPreferenceLists(),
+ int minActiveReplica = DelayedRebalanceUtil.getMinActiveReplica(ResourceConfig
+ .mergeIdealStateWithResourceConfig(clusterData.getResourceConfig(resourceName),
+ currentIdealState), currentIdealState, numReplica);
+ Map<String, List<String>> finalPreferenceLists =
+ DelayedRebalanceUtil.getFinalDelayedMapping(newActiveIdealState.getPreferenceLists(),
newIdealState.getPreferenceLists(), enabledLiveInstances,
Math.min(minActiveReplica, numReplica));
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
index 41c43d6..e2267cf 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/model/ClusterModelProvider.java
@@ -472,14 +472,15 @@ public class ClusterModelProvider {
}
Map<String, Integer> stateCountMap =
def.getStateCountMap(activeFaultZoneCount, is.getReplicaCount(assignableNodes.size()));
- mergeIdealStateWithResourceConfig(resourceConfig, is);
+ ResourceConfig mergedResourceConfig =
+ ResourceConfig.mergeIdealStateWithResourceConfig(resourceConfig, is);
Set<AssignableReplica> replicas = new HashSet<>();
for (String partition : is.getPartitionSet()) {
for (Map.Entry<String, Integer> entry : stateCountMap.entrySet()) {
String state = entry.getKey();
for (int i = 0; i < entry.getValue(); i++) {
- replicas.add(new AssignableReplica(clusterConfig, resourceConfig, partition, state,
- def.getStatePriorityMap().get(state)));
+ replicas.add(new AssignableReplica(clusterConfig, mergedResourceConfig, partition, state,
+ def.getStatePriorityMap().get(state)));
}
}
}
@@ -488,32 +489,6 @@ public class ClusterModelProvider {
}
/**
- * For backward compatibility, propagate the critical simple fields from the IdealState to
- * the Resource Config.
- * Eventually, Resource Config should be the only metadata node that contains the required information.
- */
- private static void mergeIdealStateWithResourceConfig(ResourceConfig resourceConfig,
- final IdealState idealState) {
- // Note that the config fields get updated in this method shall be fully compatible with ones in the IdealState.
- // 1. The fields shall have exactly the same meaning.
- // 2. The value shall be exactly compatible, no additional calculation involved.
- // 3. Resource Config items have a high priority.
- // This is to ensure the resource config is not polluted after the merge.
- if (null == resourceConfig.getRecord()
- .getSimpleField(ResourceConfig.ResourceConfigProperty.INSTANCE_GROUP_TAG.name())) {
- resourceConfig.getRecord()
- .setSimpleField(ResourceConfig.ResourceConfigProperty.INSTANCE_GROUP_TAG.name(),
- idealState.getInstanceGroupTag());
- }
- if (null == resourceConfig.getRecord()
- .getSimpleField(ResourceConfig.ResourceConfigProperty.MAX_PARTITIONS_PER_INSTANCE.name())) {
- resourceConfig.getRecord()
- .setIntField(ResourceConfig.ResourceConfigProperty.MAX_PARTITIONS_PER_INSTANCE.name(),
- idealState.getMaxPartitionsPerInstance());
- }
- }
-
- /**
* @return A map containing the assignments for each fault zone. <fault zone, <resource, set of partitions>>
*/
private static Map<String, Map<String, Set<String>>> mapAssignmentToFaultZone(
diff --git a/helix-core/src/main/java/org/apache/helix/model/ResourceConfig.java b/helix-core/src/main/java/org/apache/helix/model/ResourceConfig.java
index f4d8b2b..5e03695 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ResourceConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ResourceConfig.java
@@ -27,10 +27,10 @@ import java.util.Map;
import java.util.TreeMap;
import org.apache.helix.HelixProperty;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.api.config.HelixConfigProperty;
import org.apache.helix.api.config.RebalanceConfig;
import org.apache.helix.api.config.StateTransitionTimeoutConfig;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
@@ -545,7 +545,6 @@ public class ResourceConfig extends HelixProperty {
return true;
}
-
public static class Builder {
private String _resourceId;
private Boolean _monitorDisabled;
@@ -842,5 +841,72 @@ public class ResourceConfig extends HelixProperty {
_mapFields, _p2pMessageEnabled, _partitionCapacityMap);
}
}
+
+ /**
+ * For backward compatibility, propagate the critical simple fields from the IdealState to
+ * the Resource Config.
+ * Eventually, Resource Config should be the only metadata node that contains the required information.
+ *
+ * Note that the config fields get updated in this method shall be fully compatible with ones in the IdealState.
+ * 1. The fields shall have exactly the same meaning.
+ * 2. The value shall be fully compatible, no additional calculation involved.
+ * 3. Resource Config items have a high priority.
+ */
+ public static ResourceConfig mergeIdealStateWithResourceConfig(
+ final ResourceConfig resourceConfig, final IdealState idealState) {
+ if (idealState == null) {
+ return resourceConfig;
+ }
+ ResourceConfig mergedResourceConfig;
+ if (resourceConfig != null) {
+ if (!resourceConfig.getResourceName().equals(idealState.getResourceName())) {
+ throw new IllegalArgumentException(String.format(
+ "Cannot merge the IdealState of resource %s with the ResourceConfig of resource %s",
+ resourceConfig.getResourceName(), idealState.getResourceName()));
+ }
+ // Copy the resource config to avoid the original value being modified unexpectedly.
+ mergedResourceConfig = new ResourceConfig(resourceConfig.getRecord());
+ } else {
+ // If no resource config specified, construct one based on the Idealstate.
+ mergedResourceConfig = new ResourceConfig(idealState.getResourceName());
+ }
+ // Fill the compatible Idealstate fields to the ResourceConfig if possible.
+ ZNRecord mergedZNRecord = mergedResourceConfig.getRecord();
+ mergedZNRecord
+ .setSimpleFieldIfAbsent(ResourceConfig.ResourceConfigProperty.INSTANCE_GROUP_TAG.name(),
+ idealState.getInstanceGroupTag());
+ mergedZNRecord.setIntFieldIfAbsent(
+ ResourceConfig.ResourceConfigProperty.MAX_PARTITIONS_PER_INSTANCE.name(),
+ idealState.getMaxPartitionsPerInstance());
+ mergedZNRecord.setIntFieldIfAbsent(ResourceConfigProperty.NUM_PARTITIONS.name(),
+ idealState.getNumPartitions());
+ mergedZNRecord
+ .setSimpleFieldIfAbsent(ResourceConfig.ResourceConfigProperty.STATE_MODEL_DEF_REF.name(),
+ idealState.getStateModelDefRef());
+ mergedZNRecord.setSimpleFieldIfAbsent(
+ ResourceConfig.ResourceConfigProperty.STATE_MODEL_FACTORY_NAME.name(),
+ idealState.getStateModelFactoryName());
+ mergedZNRecord.setSimpleFieldIfAbsent(ResourceConfig.ResourceConfigProperty.REPLICAS.name(),
+ idealState.getReplicas());
+ mergedZNRecord
+ .setIntFieldIfAbsent(ResourceConfig.ResourceConfigProperty.MIN_ACTIVE_REPLICAS.name(),
+ idealState.getMinActiveReplicas());
+ mergedZNRecord
+ .setBooleanFieldIfAbsent(ResourceConfig.ResourceConfigProperty.HELIX_ENABLED.name(),
+ idealState.isEnabled());
+ mergedZNRecord
+ .setSimpleFieldIfAbsent(ResourceConfig.ResourceConfigProperty.RESOURCE_GROUP_NAME.name(),
+ idealState.getResourceGroupName());
+ mergedZNRecord
+ .setSimpleFieldIfAbsent(ResourceConfig.ResourceConfigProperty.RESOURCE_TYPE.name(),
+ idealState.getResourceType());
+ mergedZNRecord.setBooleanFieldIfAbsent(
+ ResourceConfig.ResourceConfigProperty.EXTERNAL_VIEW_DISABLED.name(),
+ idealState.isExternalViewDisabled());
+ mergedZNRecord.setBooleanFieldIfAbsent(
+ ResourceConfig.ResourceConfigProperty.DELAY_REBALANCE_ENABLED.name(),
+ idealState.isDelayRebalanceEnabled());
+ return mergedResourceConfig;
+ }
}
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestResourceConfig.java b/helix-core/src/test/java/org/apache/helix/model/TestResourceConfig.java
index 2e13c3e..b997878 100644
--- a/helix-core/src/test/java/org/apache/helix/model/TestResourceConfig.java
+++ b/helix-core/src/test/java/org/apache/helix/model/TestResourceConfig.java
@@ -19,17 +19,17 @@ package org.apache.helix.model;
* under the License.
*/
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
import com.google.common.collect.ImmutableMap;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.codehaus.jackson.map.ObjectMapper;
import org.testng.Assert;
import org.testng.annotations.Test;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
public class TestResourceConfig {
private static final ObjectMapper _objectMapper = new ObjectMapper();
@@ -183,4 +183,89 @@ public class TestResourceConfig {
builder.build();
}
+
+ @Test
+ public void testMergeWithIdealState() {
+ // Test failure case
+ ResourceConfig testConfig = new ResourceConfig("testResource");
+ IdealState testIdealState = new IdealState("DifferentState");
+ try {
+ ResourceConfig.mergeIdealStateWithResourceConfig(testConfig, testIdealState);
+ Assert.fail("Should not be able merge with a IdealState of different resource.");
+ } catch (IllegalArgumentException ex) {
+ // expected
+ }
+ testIdealState = new IdealState("testResource");
+ testIdealState.setInstanceGroupTag("testISGroup");
+ testIdealState.setMaxPartitionsPerInstance(1);
+ testIdealState.setNumPartitions(1);
+ testIdealState.setStateModelDefRef("testISDef");
+ testIdealState.setStateModelFactoryName("testISFactory");
+ testIdealState.setReplicas("3");
+ testIdealState.setMinActiveReplicas(1);
+ testIdealState.enable(true);
+ testIdealState.setResourceGroupName("testISGroup");
+ testIdealState.setResourceType("ISType");
+ testIdealState.setDisableExternalView(false);
+ testIdealState.setDelayRebalanceEnabled(true);
+ // Test IdealState info overriding the empty config fields.
+ ResourceConfig mergedResourceConfig =
+ ResourceConfig.mergeIdealStateWithResourceConfig(null, testIdealState);
+ Assert.assertEquals(mergedResourceConfig.getInstanceGroupTag(),
+ testIdealState.getInstanceGroupTag());
+ Assert.assertEquals(mergedResourceConfig.getMaxPartitionsPerInstance(),
+ testIdealState.getMaxPartitionsPerInstance());
+ Assert.assertEquals(mergedResourceConfig.getNumPartitions(), testIdealState.getNumPartitions());
+ Assert.assertEquals(mergedResourceConfig.getStateModelDefRef(),
+ testIdealState.getStateModelDefRef());
+ Assert.assertEquals(mergedResourceConfig.getStateModelFactoryName(),
+ testIdealState.getStateModelFactoryName());
+ Assert.assertEquals(mergedResourceConfig.getNumReplica(), testIdealState.getReplicas());
+ Assert.assertEquals(mergedResourceConfig.getMinActiveReplica(),
+ testIdealState.getMinActiveReplicas());
+ Assert
+ .assertEquals(mergedResourceConfig.isEnabled().booleanValue(), testIdealState.isEnabled());
+ Assert.assertEquals(mergedResourceConfig.getResourceGroupName(),
+ testIdealState.getResourceGroupName());
+ Assert.assertEquals(mergedResourceConfig.getResourceType(), testIdealState.getResourceType());
+ Assert.assertEquals(mergedResourceConfig.isExternalViewDisabled().booleanValue(),
+ testIdealState.isExternalViewDisabled());
+ Assert.assertEquals(Boolean.valueOf(mergedResourceConfig
+ .getSimpleConfig(ResourceConfig.ResourceConfigProperty.DELAY_REBALANCE_ENABLED.name()))
+ .booleanValue(), testIdealState.isDelayRebalanceEnabled());
+ // Test priority, Resource Config field has higher priority.
+ ResourceConfig.Builder configBuilder = new ResourceConfig.Builder("testResource");
+ configBuilder.setInstanceGroupTag("testRCGroup");
+ configBuilder.setMaxPartitionsPerInstance(2);
+ configBuilder.setNumPartitions(2);
+ configBuilder.setStateModelDefRef("testRCDef");
+ configBuilder.setStateModelFactoryName("testRCFactory");
+ configBuilder.setNumReplica("4");
+ configBuilder.setMinActiveReplica(2);
+ configBuilder.setHelixEnabled(false);
+ configBuilder.setResourceGroupName("testRCGroup");
+ configBuilder.setResourceType("RCType");
+ configBuilder.setExternalViewDisabled(true);
+ testConfig = configBuilder.build();
+ mergedResourceConfig =
+ ResourceConfig.mergeIdealStateWithResourceConfig(testConfig, testIdealState);
+ Assert
+ .assertEquals(mergedResourceConfig.getInstanceGroupTag(), testConfig.getInstanceGroupTag());
+ Assert.assertEquals(mergedResourceConfig.getMaxPartitionsPerInstance(),
+ testConfig.getMaxPartitionsPerInstance());
+ Assert.assertEquals(mergedResourceConfig.getNumPartitions(), testConfig.getNumPartitions());
+ Assert
+ .assertEquals(mergedResourceConfig.getStateModelDefRef(), testConfig.getStateModelDefRef());
+ Assert.assertEquals(mergedResourceConfig.getStateModelFactoryName(),
+ testConfig.getStateModelFactoryName());
+ Assert.assertEquals(mergedResourceConfig.getNumReplica(), testConfig.getNumReplica());
+ Assert
+ .assertEquals(mergedResourceConfig.getMinActiveReplica(), testConfig.getMinActiveReplica());
+ Assert.assertEquals(mergedResourceConfig.isEnabled(), testConfig.isEnabled());
+ Assert.assertEquals(mergedResourceConfig.getResourceGroupName(),
+ testConfig.getResourceGroupName());
+ Assert.assertEquals(mergedResourceConfig.getResourceType(), testConfig.getResourceType());
+ Assert.assertEquals(mergedResourceConfig.isExternalViewDisabled(),
+ testConfig.isExternalViewDisabled());
+ }
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/datamodel/ZNRecord.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/datamodel/ZNRecord.java
index 38e4788..6e32ff4 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/datamodel/ZNRecord.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/datamodel/ZNRecord.java
@@ -218,6 +218,16 @@ public class ZNRecord {
simpleFields.put(k, v);
}
+ /**
+ * Set a value with the input key if the key is absent.
+ * @param k
+ * @param v
+ */
+ @JsonProperty
+ public void setSimpleFieldIfAbsent(String k, String v) {
+ simpleFields.putIfAbsent(k, v);
+ }
+
@JsonProperty
public String getId() {
return id;
@@ -325,6 +335,15 @@ public class ZNRecord {
}
/**
+ * Set a single simple int field with the input key if the key is absent.
+ * @param k
+ * @param v
+ */
+ public void setIntFieldIfAbsent(String k, int v) {
+ setSimpleFieldIfAbsent(k, Integer.toString(v));
+ }
+
+ /**
* Get a single int field
* @param k
* @param defaultValue
@@ -409,6 +428,17 @@ public class ZNRecord {
}
/**
+ * Set a single simple boolean field with the input key if the key is absent.
+ *
+ * @param k
+ * @param v
+ */
+ @JsonProperty
+ public void setBooleanFieldIfAbsent(String k, boolean v) {
+ setSimpleFieldIfAbsent(k, Boolean.toString(v));
+ }
+
+ /**
* Get a single boolean field
* @param k
* @param defaultValue