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 <[email protected]>
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
