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/04/14 23:27:46 UTC
[helix] branch master updated: Fix unexpceted partition movements
in the CrushEd strategy. (#941)
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 8d6586b Fix unexpceted partition movements in the CrushEd strategy. (#941)
8d6586b is described below
commit 8d6586bc4ec8e74d6e4db65d60c9f75ba012588b
Author: Jiajun Wang <18...@users.noreply.github.com>
AuthorDate: Tue Apr 14 16:27:36 2020 -0700
Fix unexpceted partition movements in the CrushEd strategy. (#941)
This is a workaround fix to ensure backward compatibility. An additional cache map is used to keep the stable partition list so as to remove the randomness in the algorithm input.
Note that the right fix would be cleaner that we sort the list inside the strategy class. However, that will also change all existing cluster assignments in production.
---
.../ResourceControllerDataProvider.java | 51 +++++++++++-
.../controller/rebalancer/AbstractRebalancer.java | 21 ++++-
.../controller/rebalancer/AutoRebalancer.java | 4 +-
.../rebalancer/DelayedAutoRebalancer.java | 2 +-
.../TestResourceControllerDataProvider.java | 91 ++++++++++++++++++++++
5 files changed, 163 insertions(+), 6 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ResourceControllerDataProvider.java b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ResourceControllerDataProvider.java
index 74a93c7..e8e5738 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ResourceControllerDataProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ResourceControllerDataProvider.java
@@ -19,24 +19,26 @@ package org.apache.helix.controller.dataproviders;
* under the License.
*/
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-
import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.common.caches.AbstractDataCache;
import org.apache.helix.common.caches.PropertyCache;
import org.apache.helix.controller.LogUtil;
import org.apache.helix.controller.pipeline.Pipeline;
import org.apache.helix.controller.stages.MissingTopStateRecord;
import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
import org.apache.helix.model.ResourceAssignment;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +70,15 @@ public class ResourceControllerDataProvider extends BaseControllerDataProvider {
// Maintain a set of all ChangeTypes for change detection
private Set<HelixConstants.ChangeType> _refreshedChangeTypes;
+ // CrushEd strategy needs to have a stable partition list input. So this cached list persist the
+ // previous seen partition lists. If the members in a list are not modified, the old list will be
+ // used in the algorithm to ensure it calculates the same result.
+ // Refer to https://github.com/apache/helix/issues/940.
+ // TODO: Sorting the partition list in CrushEd strategy instead of using the cache to reduce
+ // TODO: dependency. Note that this will change the cluster partition assignment and potentially
+ // TODO: cause shuffling. So it is not backward compatible.
+ private final Map<String, List<String>> _stablePartitionListCache = new HashMap<>();
+
public ResourceControllerDataProvider() {
this(AbstractDataCache.UNKNOWN_CLUSTER);
}
@@ -132,6 +143,13 @@ public class ResourceControllerDataProvider extends BaseControllerDataProvider {
// Refresh resource controller specific property caches
refreshExternalViews(accessor);
refreshTargetExternalViews(accessor);
+
+ // This is part of the backward compatible workaround to fix
+ // https://github.com/apache/helix/issues/940.
+ // TODO: remove the workaround once we are able to apply the simple fix without majorly
+ // TODO: impacting user's clusters.
+ refreshStablePartitionList(getIdealStates());
+
LogUtil.logInfo(logger, getClusterEventId(), String.format(
"END: ResourceControllerDataProvider.refresh() for cluster %s, started at %d took %d for %s pipeline",
getClusterName(), startTime, System.currentTimeMillis() - startTime, getPipelineName()));
@@ -294,4 +312,33 @@ public class ResourceControllerDataProvider extends BaseControllerDataProvider {
_missingTopStateMap.clear();
_lastTopStateLocationMap.clear();
}
+
+ /**
+ * This is for a backward compatible workaround to fix https://github.com/apache/helix/issues/940.
+ *
+ * @param resourceName
+ * @return the cached stable partition list of the specified resource. If no such cached item,
+ * return null.
+ */
+ public List<String> getStablePartitionList(String resourceName) {
+ return _stablePartitionListCache.get(resourceName);
+ }
+
+ /**
+ * Refresh the stable partition list cache items and remove the non-exist resources' cache.
+ * This is for a backward compatible workaround to fix https://github.com/apache/helix/issues/940.
+ *
+ * @param idealStateMap
+ */
+ final void refreshStablePartitionList(Map<String, IdealState> idealStateMap) {
+ _stablePartitionListCache.keySet().retainAll(idealStateMap.keySet());
+ for (String resourceName : idealStateMap.keySet()) {
+ Set<String> newPartitionSet = idealStateMap.get(resourceName).getPartitionSet();
+ List<String> cachedPartitionList = getStablePartitionList(resourceName);
+ if (cachedPartitionList == null || cachedPartitionList.size() != newPartitionSet.size()
+ || !newPartitionSet.containsAll(cachedPartitionList)) {
+ _stablePartitionListCache.put(resourceName, new ArrayList<>(newPartitionSet));
+ }
+ }
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AbstractRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AbstractRebalancer.java
index 17343e6..9416c7e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AbstractRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AbstractRebalancer.java
@@ -32,7 +32,6 @@ import java.util.Set;
import org.apache.helix.HelixDefinedState;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.rebalancer.internal.MappingCalculator;
@@ -46,6 +45,7 @@ import org.apache.helix.model.Resource;
import org.apache.helix.model.ResourceAssignment;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.util.HelixUtil;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -417,4 +417,23 @@ public abstract class AbstractRebalancer<T extends BaseControllerDataProvider> i
return p1.compareTo(p2);
}
}
+
+ // This is for a backward compatible workaround to fix
+ // https://github.com/apache/helix/issues/940.
+ // TODO: remove the workaround once we are able to apply the simple fix without majorly
+ // TODO: impacting user's clusters.
+ protected List<String> getStablePartitionList(ResourceControllerDataProvider clusterData,
+ IdealState currentIdealState) {
+ List<String> partitions =
+ clusterData.getStablePartitionList(currentIdealState.getResourceName());
+ if (partitions == null) {
+ Set<String> currentPartitionSet = currentIdealState.getPartitionSet();
+ // In theory, the cached stable partition list must have contains all items in the current
+ // partition set. Add one more check to avoid any intermediate change that modifies the list.
+ LOG.warn("The current partition set {} has not been cached in the stable partition list. "
+ + "Use the IdealState partition set directly.", currentPartitionSet.toString());
+ partitions = new ArrayList<>(currentPartitionSet);
+ }
+ return partitions;
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
index f74f1eb..cb323a3 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
@@ -28,13 +28,13 @@ import java.util.Map;
import java.util.Set;
import org.apache.helix.HelixException;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.stages.CurrentStateOutput;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.IdealState.RebalanceMode;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,7 +64,7 @@ public class AutoRebalancer extends AbstractRebalancer<ResourceControllerDataPro
LOG.info("Computing IdealState for " + resourceName);
- List<String> partitions = new ArrayList<>(currentIdealState.getPartitionSet());
+ List<String> partitions = getStablePartitionList(clusterData, currentIdealState);
String stateModelName = currentIdealState.getStateModelDefRef();
StateModelDefinition stateModelDef = clusterData.getStateModelDef(stateModelName);
if (stateModelDef == null) {
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 1997cfa..53c1225 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
@@ -66,7 +66,7 @@ public class DelayedAutoRebalancer extends AbstractRebalancer<ResourceController
LOG.info("Computing IdealState for " + resourceName);
- List<String> allPartitions = new ArrayList<>(currentIdealState.getPartitionSet());
+ List<String> allPartitions = getStablePartitionList(clusterData, currentIdealState);
if (allPartitions.size() == 0) {
LOG.info("Partition count is 0 for resource " + resourceName
+ ", stop calculate ideal mapping for the resource.");
diff --git a/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestResourceControllerDataProvider.java b/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestResourceControllerDataProvider.java
new file mode 100644
index 0000000..ebf37d4
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestResourceControllerDataProvider.java
@@ -0,0 +1,91 @@
+package org.apache.helix.controller.dataproviders;
+
+/*
+ * 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.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.helix.model.IdealState;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestResourceControllerDataProvider {
+ @Test
+ public void testStablePartitionListCache() {
+ String resourceName = "TestResource";
+ Set<String> partitionSetA = ImmutableSet.of("Partiton1", "Partiton2", "Partiton3");
+ Set<String> partitionSetB = ImmutableSet.of("Partiton1", "Partiton2", "Partiton4");
+ Map<String, IdealState> idealStateMap = new HashMap<>();
+ IdealState is = mock(IdealState.class);
+ when(is.getPartitionSet()).thenReturn(partitionSetA);
+ when(is.getResourceName()).thenReturn(resourceName);
+ idealStateMap.put(resourceName, is);
+
+ ResourceControllerDataProvider dataProvider = new ResourceControllerDataProvider();
+
+ List<String> cachedPartitionList =
+ dataProvider.getStablePartitionList(resourceName);
+ Assert.assertTrue(cachedPartitionList == null);
+
+ // 1. Test refresh and get stable list
+ dataProvider.refreshStablePartitionList(idealStateMap);
+ List<String> cachedPartitionListA =
+ dataProvider.getStablePartitionList(resourceName);
+ Assert.assertTrue(cachedPartitionListA.size() == partitionSetA.size() && cachedPartitionListA
+ .containsAll(partitionSetA));
+
+ Set<String> partitionSetAWithDifferentOrder = new LinkedHashSet<>();
+ partitionSetAWithDifferentOrder.add("Partiton3");
+ partitionSetAWithDifferentOrder.add("Partiton2");
+ partitionSetAWithDifferentOrder.add("Partiton1");
+ // Verify that iterating this list will generate a different result
+ List<String> tmpPartitionList = new ArrayList<>(partitionSetAWithDifferentOrder);
+ Assert.assertFalse(cachedPartitionListA.equals(tmpPartitionList));
+ // Verify that the cached stable partition list still return a list with the same order even
+ // after refresh call.
+ when(is.getPartitionSet()).thenReturn(partitionSetAWithDifferentOrder);
+ dataProvider.refreshStablePartitionList(idealStateMap);
+ Assert
+ .assertTrue(dataProvider.getStablePartitionList(resourceName).equals(cachedPartitionListA));
+
+ // 2. Test update the cache if items in the list have been changed.
+ when(is.getPartitionSet()).thenReturn(partitionSetB);
+ dataProvider.refreshStablePartitionList(idealStateMap);
+ List<String> cachedPartitionListB =
+ dataProvider.getStablePartitionList(resourceName);
+ Assert.assertTrue(cachedPartitionListB.size() == partitionSetB.size() && cachedPartitionListB
+ .containsAll(partitionSetB));
+
+ // 3. Test removing item from the cache once the IdealState has been removed
+ idealStateMap.clear();
+ dataProvider.refreshStablePartitionList(idealStateMap);
+ // Now, since the cache has been cleaned, the get will return different order.
+ Assert.assertTrue(
+ dataProvider.getStablePartitionList(resourceName) == null);
+ }
+}