You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by xy...@apache.org on 2024/02/13 20:54:25 UTC
(helix) branch master updated: Allow Evac + Add swap capability in HelixAdmin and SWAP behavior changes for offline or disabled SWAP_OUT node (#2755)
This is an automated email from the ASF dual-hosted git repository.
xyuanlu 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 908ff6e63 Allow Evac + Add swap capability in HelixAdmin and SWAP behavior changes for offline or disabled SWAP_OUT node (#2755)
908ff6e63 is described below
commit 908ff6e63578d12c9d5dd0a0dad58a5e065df670
Author: Zachary Pinto <za...@linkedin.com>
AuthorDate: Tue Feb 13 12:54:18 2024 -0800
Allow Evac + Add swap capability in HelixAdmin and SWAP behavior changes for offline or disabled SWAP_OUT node (#2755)
*Modify the behavior of swap to continue bootstrapping second or top state on the swap in node regardless of if the swap out node is disabled or offline. We don't want either of those two cases to nullify the work already done during the swap. Modify HelixAdmin to allow evacuation to be used with adding a new instance with the same logicalId and allow for swap to complete when swap out instance is offline or disabled.
---
.../src/main/java/org/apache/helix/HelixAdmin.java | 3 +-
.../main/java/org/apache/helix/HelixManager.java | 3 +-
.../stages/BestPossibleStateCalcStage.java | 50 +++--
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 179 ++++++++--------
.../rebalancer/TestInstanceOperation.java | 231 ++++++++++++++-------
.../java/org/apache/helix/mock/MockHelixAdmin.java | 8 +-
.../resources/helix/PerInstanceAccessor.java | 5 +-
7 files changed, 286 insertions(+), 193 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
index f53b886e2..021332be7 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
@@ -771,10 +771,11 @@ public interface HelixAdmin {
*
* @param clusterName The cluster name
* @param instanceName The instance that is being swapped out or swapped in
+ * @param forceComplete Whether to force complete the swap without checking if it is ready
* @return True if the swap is ready to be completed and was completed successfully, false
* otherwise.
*/
- boolean completeSwapIfPossible(String clusterName, String instanceName);
+ boolean completeSwapIfPossible(String clusterName, String instanceName, boolean forceComplete);
/**
* Return if instance is ready for preparing joining cluster. The instance should have no current state,
diff --git a/helix-core/src/main/java/org/apache/helix/HelixManager.java b/helix-core/src/main/java/org/apache/helix/HelixManager.java
index 4ce3ff9ad..b5c2a6448 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixManager.java
@@ -241,8 +241,7 @@ public interface HelixManager {
}
/**
-
- * @see CustomizedStateRootChangeListener#onCustomizedStateRootChange(String, NotificationContext)
+ * @see CustomizedStateRootChangeListener#(String, List, NotificationContext)
* @param listener
* @param instanceName
*/
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 36b4583ac..2a6f9644e 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
@@ -138,14 +138,20 @@ public class BestPossibleStateCalcStage extends AbstractBaseStage {
// 2. Get all enabled and live SWAP_IN instances in the cluster.
Set<String> liveSwapInInstances = cache.getLiveSwapInInstanceNames();
Set<String> enabledSwapInInstances = cache.getEnabledSwapInInstanceNames();
- // 3. For each SWAP_OUT instance in any of the preferenceLists, add the corresponding SWAP_IN instance to the end.
- // Skipping this when there are not SWAP_IN instances that are alive will reduce computation time.
+ // 3. For each SWAP_OUT instance in any of the preferenceLists, add the corresponding SWAP_IN instance to
+ // the stateMap with the correct state.
+ // Skipping this when there are no SWAP_IN instances that are alive will reduce computation time.
if (!liveSwapInInstances.isEmpty() && !cache.isMaintenanceModeEnabled()) {
resourceMap.forEach((resourceName, resource) -> {
StateModelDefinition stateModelDef = cache.getStateModelDef(resource.getStateModelDefRef());
bestPossibleStateOutput.getResourceStatesMap().get(resourceName).getStateMap()
.forEach((partition, stateMap) -> {
- Set<String> commonInstances = new HashSet<>(stateMap.keySet());
+ // We use the preferenceList for the case where the swapOutInstance goes offline.
+ // We do not want to drop the replicas that may have been bootstrapped on the swapInInstance
+ // in the case that the swapOutInstance goes offline and no longer has an entry in the stateMap.
+ Set<String> commonInstances = new HashSet<>(
+ bestPossibleStateOutput.getPreferenceList(resourceName,
+ partition.getPartitionName()));
commonInstances.retainAll(swapOutToSwapInInstancePairs.keySet());
commonInstances.forEach(swapOutInstance -> {
@@ -165,37 +171,39 @@ public class BestPossibleStateCalcStage extends AbstractBaseStage {
}
// If the swap-in node is live and enabled, do assignment with the following logic:
- // 1. If the swap-out instance's replica is a topState, set the swap-in instance's replica
- // to the topState if the StateModel allows for another replica with the topState to be added.
- // Otherwise, set the swap-in instance's replica to the secondTopState.
- // 2. If the swap-out instance's replica is a secondTopState, set the swap-in instance's replica
+ // 1. If the swap-out instance's replica is a secondTopState, set the swap-in instance's replica
// to the same secondTopState.
- if (stateMap.get(swapOutInstance).equals(stateModelDef.getTopState())) {
-
+ // 2. If the swap-out instance's replica is any other state and is in the preferenceList,
+ // set the swap-in instance's replica to the topState if the StateModel allows another to be added.
+ // If not, set the swap-in instance's replica to the secondTopState.
+ // We can make this assumption because if there is assignment to the swapOutInstance, it must be either
+ // a topState or a secondTopState.
+ if (stateMap.containsKey(swapOutInstance) && stateModelDef.getSecondTopStates()
+ .contains(stateMap.get(swapOutInstance))) {
+ // If the swap-out instance's replica is a secondTopState, set the swap-in instance's replica
+ // to the same secondTopState.
+ stateMap.put(swapOutToSwapInInstancePairs.get(swapOutInstance),
+ stateMap.get(swapOutInstance));
+ } else {
+ // If the swap-out instance's replica is any other state in the stateMap or not present in the
+ // stateMap, set the swap-in instance's replica to the topState if the StateModel allows another
+ // to be added. If not, set the swap-in to the secondTopState.
String topStateCount =
stateModelDef.getNumInstancesPerState(stateModelDef.getTopState());
if (topStateCount.equals(
StateModelDefinition.STATE_REPLICA_COUNT_ALL_CANDIDATE_NODES)
|| topStateCount.equals(
StateModelDefinition.STATE_REPLICA_COUNT_ALL_REPLICAS)) {
- // If the swap-out instance's replica is a topState and the StateModel allows for
- // another replica with the topState to be added, set the swap-in instance's replica
- // to the topState.
+ // If the StateModel allows for another replica with the topState to be added,
+ // set the swap-in instance's replica to the topState.
stateMap.put(swapOutToSwapInInstancePairs.get(swapOutInstance),
stateModelDef.getTopState());
} else {
- // If the swap-out instance's replica is a topState and the StateModel does not allow for
- // another replica with the topState to be added, set the swap-in instance's replica
- // to the secondTopState.
+ // If StateModel does not allow another topState replica to be
+ // added, set the swap-in instance's replica to the secondTopState.
stateMap.put(swapOutToSwapInInstancePairs.get(swapOutInstance),
stateModelDef.getSecondTopStates().iterator().next());
}
- } else if (stateModelDef.getSecondTopStates()
- .contains(stateMap.get(swapOutInstance))) {
- // If the swap-out instance's replica is a secondTopState, set the swap-in instance's replica
- // to the same secondTopState.
- stateMap.put(swapOutToSwapInInstancePairs.get(swapOutInstance),
- stateMap.get(swapOutInstance));
}
});
});
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index c5303cdd2..d52967f5c 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -242,61 +242,68 @@ public class ZKHelixAdmin implements HelixAdmin {
.getInstanceName() + " already have the same logicalId: " + toAddInstanceLogicalId
+ "; therefore, " + nodeId + " cannot be added to the cluster.");
} else if (foundInstanceConfigsWithMatchingLogicalId.size() == 1) {
- // If there is only one instance with the same logicalId, we can infer that the intended behaviour
- // is to SWAP_IN.
-
- // If the InstanceOperation is unset, we will set it to SWAP_IN.
- if (!instanceConfig.getInstanceOperation()
- .equals(InstanceConstants.InstanceOperation.SWAP_IN.name())) {
- instanceConfig.setInstanceOperation(InstanceConstants.InstanceOperation.SWAP_IN);
- }
-
- // If the existing instance with the same logicalId does not have InstanceOperation set to SWAP_OUT and this instance
- // is attempting to join as enabled, we cannot add this instance.
- if (instanceConfig.getInstanceEnabled() && !foundInstanceConfigsWithMatchingLogicalId.get(0)
- .getInstanceOperation()
+ // If there is only one instance with the same logicalId,
+ // we can infer that the intended behaviour is to SWAP_IN or EVACUATE + ADD.
+ if (foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceOperation()
.equals(InstanceConstants.InstanceOperation.SWAP_OUT.name())) {
- throw new HelixException(
- "Instance can only be added if the exising instance sharing the same logicalId has InstanceOperation"
- + " set to " + InstanceConstants.InstanceOperation.SWAP_OUT.name()
- + " and this instance has InstanceOperation set to "
- + InstanceConstants.InstanceOperation.SWAP_IN.name() + ". " + "Existing instance: "
- + foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceName()
- + " has InstanceOperation: "
- + foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceOperation()
- + " and this instance: " + nodeId + " has InstanceOperation: "
- + instanceConfig.getInstanceOperation());
- }
+ // If the existing instance with the same logicalId has SWAP_OUT InstanceOperation
- // If the existing instance with the same logicalId is not in the same FAULT_ZONE as this instance, we cannot
- // add this instance.
- if (!foundInstanceConfigsWithMatchingLogicalId.get(0).getDomainAsMap()
- .containsKey(faultZoneKey) || !instanceConfig.getDomainAsMap().containsKey(faultZoneKey)
- || !foundInstanceConfigsWithMatchingLogicalId.get(0).getDomainAsMap().get(faultZoneKey)
- .equals(instanceConfig.getDomainAsMap().get(faultZoneKey))) {
- throw new HelixException(
- "Instance can only be added if the SWAP_OUT instance sharing the same logicalId is in the same FAULT_ZONE"
- + " as this instance. " + "Existing instance: "
- + foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceName()
- + " has FAULT_ZONE_TYPE: " + foundInstanceConfigsWithMatchingLogicalId.get(0)
- .getDomainAsMap().get(faultZoneKey) + " and this instance: " + nodeId
- + " has FAULT_ZONE_TYPE: " + instanceConfig.getDomainAsMap().get(faultZoneKey));
- }
+ // If the InstanceOperation is unset, we will set it to SWAP_IN.
+ if (!instanceConfig.getInstanceOperation()
+ .equals(InstanceConstants.InstanceOperation.SWAP_IN.name())) {
+ instanceConfig.setInstanceOperation(InstanceConstants.InstanceOperation.SWAP_IN);
+ }
- Map<String, Integer> foundInstanceCapacityMap =
- foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceCapacityMap().isEmpty()
- ? clusterConfig.getDefaultInstanceCapacityMap()
- : foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceCapacityMap();
- Map<String, Integer> instanceCapacityMap = instanceConfig.getInstanceCapacityMap().isEmpty()
- ? clusterConfig.getDefaultInstanceCapacityMap() : instanceConfig.getInstanceCapacityMap();
- // If the instance does not have the same capacity, we cannot add this instance.
- if (!new EqualsBuilder().append(foundInstanceCapacityMap, instanceCapacityMap).isEquals()) {
+ // If the existing instance with the same logicalId is not in the same FAULT_ZONE as this instance, we cannot
+ // add this instance.
+ if (!foundInstanceConfigsWithMatchingLogicalId.get(0).getDomainAsMap()
+ .containsKey(faultZoneKey) || !instanceConfig.getDomainAsMap().containsKey(faultZoneKey)
+ || !foundInstanceConfigsWithMatchingLogicalId.get(0).getDomainAsMap().get(faultZoneKey)
+ .equals(instanceConfig.getDomainAsMap().get(faultZoneKey))) {
+ throw new HelixException(
+ "Instance can only be added if the SWAP_OUT instance sharing the same logicalId is in the same FAULT_ZONE"
+ + " as this instance. " + "Existing instance: "
+ + foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceName()
+ + " has FAULT_ZONE_TYPE: " + foundInstanceConfigsWithMatchingLogicalId.get(0)
+ .getDomainAsMap().get(faultZoneKey) + " and this instance: " + nodeId
+ + " has FAULT_ZONE_TYPE: " + instanceConfig.getDomainAsMap().get(faultZoneKey));
+ }
+
+ Map<String, Integer> foundInstanceCapacityMap =
+ foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceCapacityMap().isEmpty()
+ ? clusterConfig.getDefaultInstanceCapacityMap()
+ : foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceCapacityMap();
+ Map<String, Integer> instanceCapacityMap = instanceConfig.getInstanceCapacityMap().isEmpty()
+ ? clusterConfig.getDefaultInstanceCapacityMap()
+ : instanceConfig.getInstanceCapacityMap();
+ // If the instance does not have the same capacity, we cannot add this instance.
+ if (!new EqualsBuilder().append(foundInstanceCapacityMap, instanceCapacityMap).isEquals()) {
+ throw new HelixException(
+ "Instance can only be added if the SWAP_OUT instance sharing the same logicalId has the same capacity"
+ + " as this instance. " + "Existing instance: "
+ + foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceName()
+ + " has capacity: " + foundInstanceCapacityMap + " and this instance: " + nodeId
+ + " has capacity: " + instanceCapacityMap);
+ }
+ } else if (foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceOperation()
+ .equals(InstanceConstants.InstanceOperation.EVACUATE.name())) {
+ // No need to check anything on the new node, the old node will be evacuated and the new node
+ // will be added.
+ } else {
+ // If the instanceConfig.getInstanceEnabled() is true and the existing instance with the same logicalId
+ // does not have InstanceOperation set to one of the above, we cannot add this instance.
throw new HelixException(
- "Instance can only be added if the SWAP_OUT instance sharing the same logicalId has the same capacity"
- + " as this instance. " + "Existing instance: "
+ "Instance can only be added if the exising instance sharing the same logicalId"
+ + " has InstanceOperation set to "
+ + InstanceConstants.InstanceOperation.SWAP_OUT.name()
+ + " and this instance has InstanceOperation set to "
+ + InstanceConstants.InstanceOperation.SWAP_IN.name()
+ + " or the existing instance sharing the same logicalId has Instance Operation set to "
+ + InstanceConstants.InstanceOperation.EVACUATE.name()
+ + " and this instance has InstanceOperation unset. Existing instance: "
+ foundInstanceConfigsWithMatchingLogicalId.get(0).getInstanceName()
- + " has capacity: " + foundInstanceCapacityMap + " and this instance: " + nodeId
- + " has capacity: " + instanceCapacityMap);
+ + " has InstanceOperation: " + foundInstanceConfigsWithMatchingLogicalId.get(0)
+ .getInstanceOperation());
}
} else if (!instanceConfig.getInstanceOperation().isEmpty()) {
// If there are no instances with the same logicalId, we can only add this instance if InstanceOperation
@@ -542,13 +549,15 @@ public class ZKHelixAdmin implements HelixAdmin {
&& !existingInstanceConfig.getInstanceOperation()
.equals(InstanceConstants.InstanceOperation.SWAP_IN.name())
&& !existingInstanceConfig.getInstanceOperation()
- .equals(InstanceConstants.InstanceOperation.SWAP_OUT.name()))
+ .equals(InstanceConstants.InstanceOperation.SWAP_OUT.name())
+ && !existingInstanceConfig.getInstanceOperation()
+ .equals(InstanceConstants.InstanceOperation.EVACUATE.name()))
.collect(Collectors.toList());
if (!matchingInstancesWithNonSwappingInstanceOperation.isEmpty()) {
throw new HelixException("InstanceOperation cannot be set to null for " + instanceName
+ " if there are other instances with the same logicalId in the cluster that do not have"
- + " InstanceOperation set to SWAP_IN or SWAP_OUT.");
+ + " InstanceOperation set to SWAP_IN, SWAP_OUT, or EVACUATE.");
}
}
@@ -645,18 +654,17 @@ public class ZKHelixAdmin implements HelixAdmin {
HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
- // 1. Check that both instances are alive and enabled.
+ // 1. Check that swap-in instance is live and enabled.
LiveInstance swapOutLiveInstance =
accessor.getProperty(keyBuilder.liveInstance(swapOutInstanceName));
LiveInstance swapInLiveInstance =
accessor.getProperty(keyBuilder.liveInstance(swapInInstanceName));
InstanceConfig swapOutInstanceConfig = getInstanceConfig(clusterName, swapOutInstanceName);
InstanceConfig swapInInstanceConfig = getInstanceConfig(clusterName, swapInInstanceName);
- if (swapOutLiveInstance == null || swapInLiveInstance == null
- || !swapOutInstanceConfig.getInstanceEnabled()
- || !swapInInstanceConfig.getInstanceEnabled()) {
+ if (swapInLiveInstance == null || !swapInInstanceConfig.getInstanceEnabled()) {
logger.warn(
- "SwapOutInstance {} is {} + {} and SwapInInstance {} is {} + {} for cluster {}. Swap will not complete unless both instances are ONLINE.",
+ "SwapOutInstance {} is {} + {} and SwapInInstance {} is {} + {} for cluster {}. Swap will"
+ + " not complete unless SwapInInstance instance is ENABLED and ONLINE.",
swapOutInstanceName, swapOutLiveInstance != null ? "ONLINE" : "OFFLINE",
swapOutInstanceConfig.getInstanceEnabled() ? "ENABLED" : "DISABLED", swapInInstanceName,
swapInLiveInstance != null ? "ONLINE" : "OFFLINE",
@@ -686,23 +694,38 @@ public class ZKHelixAdmin implements HelixAdmin {
List<Message> swapInMessages =
accessor.getChildValues(keyBuilder.messages(swapInInstanceName), true);
int swapInPendingMessageCount = swapInMessages != null ? swapInMessages.size() : 0;
- if (swapOutPendingMessageCount > 0 || swapInPendingMessageCount > 0) {
+ if ((swapOutLiveInstance != null && swapOutPendingMessageCount > 0)
+ || swapInPendingMessageCount > 0) {
logger.warn(
"SwapOutInstance {} has {} pending messages and SwapInInstance {} has {} pending messages for cluster {}."
- + " Swap will not complete unless both instances have no pending messages.",
+ + " Swap will not complete unless both SwapOutInstance(only when live)"
+ + " and SwapInInstance have no pending messages unless.",
swapOutInstanceName, swapOutPendingMessageCount, swapInInstanceName,
swapInPendingMessageCount, clusterName);
return false;
}
// 4. Collect a list of all partitions that have a current state on swapOutInstance
- String swapOutActiveSession = swapOutLiveInstance.getEphemeralOwner();
+ String swapOutLastActiveSession;
+ if (swapOutLiveInstance == null) {
+ // SwapOutInstance is down, try to find the last active session
+ if (swapOutSessions.size() != 1) {
+ logger.warn(
+ "SwapOutInstance {} is offline and has {} sessions for cluster {}. Swap can't be "
+ + "verified if last active session can't be determined. There should only be one session.",
+ swapOutInstanceName, swapOutSessions.size(), clusterName);
+ return false;
+ }
+ swapOutLastActiveSession = swapOutSessions.get(0);
+ } else {
+ swapOutLastActiveSession = swapOutLiveInstance.getEphemeralOwner();
+ }
String swapInActiveSession = swapInLiveInstance.getEphemeralOwner();
// Iterate over all resources with current states on the swapOutInstance
List<String> swapOutResources = baseAccessor.getChildNames(
PropertyPathBuilder.instanceCurrentState(clusterName, swapOutInstanceName,
- swapOutActiveSession), 0);
+ swapOutLastActiveSession), 0);
for (String swapOutResource : swapOutResources) {
// Get the topState and secondTopStates for the stateModelDef used by the resource.
IdealState idealState = accessor.getProperty(keyBuilder.idealStates(swapOutResource));
@@ -712,7 +735,7 @@ public class ZKHelixAdmin implements HelixAdmin {
Set<String> secondTopStates = stateModelDefinition.getSecondTopStates();
CurrentState swapOutResourceCurrentState = accessor.getProperty(
- keyBuilder.currentState(swapOutInstanceName, swapOutActiveSession, swapOutResource));
+ keyBuilder.currentState(swapOutInstanceName, swapOutLastActiveSession, swapOutResource));
CurrentState swapInResourceCurrentState = accessor.getProperty(
keyBuilder.currentState(swapInInstanceName, swapInActiveSession, swapOutResource));
@@ -731,9 +754,8 @@ public class ZKHelixAdmin implements HelixAdmin {
String swapOutPartitionState = swapOutResourceCurrentState.getState(partitionName);
String swapInPartitionState = swapInResourceCurrentState.getState(partitionName);
- // Neither instance should have any partitions in ERROR state.
- if (swapOutPartitionState.equals(HelixDefinedState.ERROR.name())
- || swapInPartitionState.equals(HelixDefinedState.ERROR.name())) {
+ // SwapInInstance should not have any partitions in ERROR state.
+ if (swapInPartitionState.equals(HelixDefinedState.ERROR.name())) {
logger.warn(
"SwapOutInstance {} has partition {} in state {} and SwapInInstance {} has partition {} in state {} for cluster {}."
+ " Swap will not complete unless both instances have no partitions in ERROR state.",
@@ -742,32 +764,18 @@ public class ZKHelixAdmin implements HelixAdmin {
return false;
}
- // When the state of a partition on a swapOut instance is in the topState, the state
- // of the partition on the swapInInstance should also be in the topState or a secondTopState.
+ // The state of the partition on the swapInInstance be in the topState or a secondTopState.
// It should be in a topState only if the state model allows multiple replicas in the topState.
// In all other cases it should be a secondTopState.
- if (swapOutPartitionState.equals(topState) && !(swapInPartitionState.equals(topState)
- || secondTopStates.contains(swapInPartitionState))) {
+ if (!(swapInPartitionState.equals(topState) || secondTopStates.contains(
+ swapInPartitionState))) {
logger.warn(
- "SwapOutInstance {} has partition {} in topState {} but SwapInInstance {} has partition {} in state {} for cluster {}."
+ "SwapOutInstance {} has partition {} in {} but SwapInInstance {} has partition {} in state {} for cluster {}."
+ " Swap will not complete unless SwapInInstance has partition in topState or secondState.",
swapOutInstanceName, partitionName, swapOutPartitionState, swapInInstanceName,
partitionName, swapInPartitionState, clusterName);
return false;
}
-
- // When the state of a partition on a swapOut instance is any other state, except ERROR, DROPPED or TopState,
- // the state of the partition on the swapInInstance should be the same.
- if (!swapOutPartitionState.equals(topState) && !swapOutPartitionState.equals(
- HelixDefinedState.DROPPED.name())
- && !swapOutPartitionState.equals(swapInPartitionState)) {
- logger.warn(
- "SwapOutInstance {} has partition {} in state {} but SwapInInstance {} has partition {} in state {} for cluster {}."
- + " Swap will not complete unless both instances have matching states.",
- swapOutInstanceName, partitionName, swapOutPartitionState, swapInInstanceName,
- partitionName, swapInPartitionState, clusterName);
- return false;
- }
}
}
@@ -803,7 +811,8 @@ public class ZKHelixAdmin implements HelixAdmin {
}
@Override
- public boolean completeSwapIfPossible(String clusterName, String instanceName) {
+ public boolean completeSwapIfPossible(String clusterName, String instanceName,
+ boolean forceComplete) {
InstanceConfig instanceConfig = getInstanceConfig(clusterName, instanceName);
if (instanceConfig == null) {
logger.warn(
@@ -826,7 +835,7 @@ public class ZKHelixAdmin implements HelixAdmin {
}
// Check if the swap is ready to be completed. If not, return false.
- if (!canCompleteSwap(clusterName, swapOutInstanceConfig.getInstanceName(),
+ if (forceComplete || !canCompleteSwap(clusterName, swapOutInstanceConfig.getInstanceName(),
swapInInstanceConfig.getInstanceName())) {
return false;
}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
index 0019ea83c..765add7fd 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
@@ -446,7 +446,7 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
}
@@ -570,7 +570,7 @@ public class TestInstanceOperation extends ZkTestBase {
// Assert completeSwapIfPossible is true
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
@@ -662,7 +662,7 @@ public class TestInstanceOperation extends ZkTestBase {
// Assert completeSwapIfPossible is true
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
@@ -734,7 +734,7 @@ public class TestInstanceOperation extends ZkTestBase {
.canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
// Assert completeSwapIfPossible is true
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
@@ -888,7 +888,7 @@ public class TestInstanceOperation extends ZkTestBase {
.canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
// Assert completeSwapIfPossible is true
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
@@ -946,34 +946,23 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
- // Validate that the SWAP_IN instance has the same partitions as the SWAP_OUT instance in second top state.
+ // Validate that the SWAP_IN instance has the same partitions in secondTopState as the SWAP_OUT instance
+ // did before being disabled.
Map<String, String> swapInInstancePartitionsAndStates =
getPartitionsAndStatesOnInstance(getEVs(), instanceToSwapInName);
- Assert.assertEquals(swapInInstancePartitionsAndStates.keySet().size(), 0);
-
- // Assert canSwapBeCompleted is false because SWAP_OUT instance is disabled.
- Assert.assertFalse(_gSetupTool.getClusterManagementTool()
- .canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
-
- // Enable the SWAP_OUT instance.
- _gSetupTool.getClusterManagementTool()
- .enableInstance(CLUSTER_NAME, instanceToSwapOutName, true);
-
- Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
-
- // Validate that the SWAP_IN instance has the same partitions the SWAP_OUT instance originally
- // had. Validate they are in second top state.
- swapInInstancePartitionsAndStates =
- getPartitionsAndStatesOnInstance(getEVs(), instanceToSwapInName);
Assert.assertTrue(
swapInInstancePartitionsAndStates.keySet().containsAll(swapOutInstanceOriginalPartitions));
Set<String> swapInInstanceStates = new HashSet<>(swapInInstancePartitionsAndStates.values());
swapInInstanceStates.removeAll(SECONDARY_STATE_SET);
Assert.assertEquals(swapInInstanceStates.size(), 0);
+ // Assert canSwapBeCompleted is false because SWAP_OUT instance is disabled.
+ Assert.assertTrue(_gSetupTool.getClusterManagementTool()
+ .canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
+
// Assert completeSwapIfPossible is true
Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
Assert.assertTrue(_clusterVerifier.verifyByPolling());
@@ -986,7 +975,125 @@ public class TestInstanceOperation extends ZkTestBase {
TIMEOUT);
}
- @Test(expectedExceptions = HelixException.class, dependsOnMethods = "testNodeSwapWithSwapOutInstanceDisabled")
+ @Test(dependsOnMethods = "testNodeSwapWithSwapOutInstanceDisabled")
+ public void testNodeSwapWithSwapOutInstanceOffline() throws Exception {
+ System.out.println(
+ "START TestInstanceOperation.testNodeSwapWithSwapOutInstanceOffline() at " + new Date(
+ System.currentTimeMillis()));
+
+ removeOfflineOrDisabledOrSwapInInstances();
+
+ // Store original EV
+ Map<String, ExternalView> originalEVs = getEVs();
+
+ Map<String, String> swapOutInstancesToSwapInInstances = new HashMap<>();
+
+ // Set instance's InstanceOperation to SWAP_OUT
+ String instanceToSwapOutName = _participants.get(0).getInstanceName();
+ InstanceConfig instanceToSwapOutInstanceConfig = _gSetupTool.getClusterManagementTool()
+ .getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName);
+ _gSetupTool.getClusterManagementTool().setInstanceOperation(CLUSTER_NAME, instanceToSwapOutName,
+ InstanceConstants.InstanceOperation.SWAP_OUT);
+
+ Assert.assertTrue(_clusterVerifier.verifyByPolling());
+
+ // Add instance with InstanceOperation set to SWAP_IN
+ String instanceToSwapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
+ swapOutInstancesToSwapInInstances.put(instanceToSwapOutName, instanceToSwapInName);
+ addParticipant(instanceToSwapInName, instanceToSwapOutInstanceConfig.getLogicalId(LOGICAL_ID),
+ instanceToSwapOutInstanceConfig.getDomainAsMap().get(ZONE),
+ InstanceConstants.InstanceOperation.SWAP_IN, true, -1);
+
+ // Kill the participant
+ _participants.get(0).syncStop();
+
+ Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
+
+ // Assert canSwapBeCompleted is true
+ Assert.assertTrue(_gSetupTool.getClusterManagementTool()
+ .canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
+
+ // Validate that the SWAP_OUT instance is in routing tables and SWAP_IN is not.
+ validateRoutingTablesInstance(getEVs(), instanceToSwapInName, false);
+
+ // Assert completeSwapIfPossible is true
+ Assert.assertTrue(_gSetupTool.getClusterManagementTool()
+ .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName, false));
+
+ Assert.assertTrue(_clusterVerifier.verifyByPolling());
+
+ // Validate that the SWAP_IN instance is now in the routing tables.
+ validateRoutingTablesInstance(getEVs(), instanceToSwapInName, true);
+
+ // Assert that SWAP_OUT instance is disabled and has no partitions assigned to it.
+ Assert.assertFalse(_gSetupTool.getClusterManagementTool()
+ .getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName).getInstanceEnabled());
+
+ // Validate that the SWAP_IN instance has the same partitions the SWAP_OUT instance had before
+ // swap was completed.
+ verifier(() -> (validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
+ Collections.emptySet(), Set.of(instanceToSwapInName))), TIMEOUT);
+ }
+
+ @Test(dependsOnMethods = "testNodeSwapWithSwapOutInstanceOffline")
+ public void testSwapEvacuateAdd() throws Exception {
+ System.out.println("START TestInstanceOperation.testSwapEvacuateAdd() at " + new Date(
+ System.currentTimeMillis()));
+ removeOfflineOrDisabledOrSwapInInstances();
+
+ // Store original EV
+ Map<String, ExternalView> originalEVs = getEVs();
+
+ Map<String, String> swapOutInstancesToSwapInInstances = new HashMap<>();
+
+ // Enter maintenance mode
+ _gSetupTool.getClusterManagementTool()
+ .manuallyEnableMaintenanceMode(CLUSTER_NAME, true, null, null);
+
+ // Set instance's InstanceOperation to EVACUATE
+ String instanceToSwapOutName = _participants.get(0).getInstanceName();
+ InstanceConfig instanceToSwapOutInstanceConfig = _gSetupTool.getClusterManagementTool()
+ .getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName);
+ _gSetupTool.getClusterManagementTool().setInstanceOperation(CLUSTER_NAME, instanceToSwapOutName,
+ InstanceConstants.InstanceOperation.EVACUATE);
+
+ // Validate that the assignment has not changed since setting the InstanceOperation to EVACUATE
+ Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
+ validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
+ Collections.emptySet(), Collections.emptySet());
+
+ // Add instance with InstanceOperation set to SWAP_IN
+ String instanceToSwapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
+ swapOutInstancesToSwapInInstances.put(instanceToSwapOutName, instanceToSwapInName);
+ addParticipant(instanceToSwapInName, instanceToSwapOutInstanceConfig.getLogicalId(LOGICAL_ID),
+ instanceToSwapOutInstanceConfig.getDomainAsMap().get(ZONE), null, true, -1);
+
+ // Exit maintenance mode
+ _gSetupTool.getClusterManagementTool()
+ .manuallyEnableMaintenanceMode(CLUSTER_NAME, false, null, null);
+
+ Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
+
+ // Validate that the SWAP_IN instance has the same partitions the SWAP_OUT instance had.
+ verifier(() -> (validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
+ Collections.emptySet(), Set.of(instanceToSwapInName))), TIMEOUT);
+
+ // Assert isEvacuateFinished is true
+ Assert.assertTrue(_gSetupTool.getClusterManagementTool()
+ .isEvacuateFinished(CLUSTER_NAME, instanceToSwapOutName));
+
+ // Disable the EVACUATE instance
+ _gSetupTool.getClusterManagementTool()
+ .enableInstance(CLUSTER_NAME, instanceToSwapOutName, false);
+
+ Assert.assertTrue(_clusterVerifier.verifyByPolling());
+
+ // Validate that dropping the instance has not changed the assignment
+ verifier(() -> (validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
+ Collections.emptySet(), Set.of(instanceToSwapInName))), TIMEOUT);
+ }
+
+ @Test(expectedExceptions = HelixException.class, dependsOnMethods = "testNodeSwapWithSwapOutInstanceOffline")
public void testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet() {
System.out.println(
"START TestInstanceOperation.testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet() at "
@@ -1053,7 +1160,7 @@ public class TestInstanceOperation extends ZkTestBase {
.setInstanceOperation(CLUSTER_NAME, instanceToSwapInName, null);
}
- @Test(dependsOnMethods = "testUnsetInstanceOperationOnSwapInWhenAlreadyUnsetOnSwapOut")
+ @Test(expectedExceptions = HelixException.class, dependsOnMethods = "testUnsetInstanceOperationOnSwapInWhenAlreadyUnsetOnSwapOut")
public void testNodeSwapAddSwapInFirst() throws Exception {
System.out.println("START TestInstanceOperation.testNodeSwapAddSwapInFirst() at " + new Date(
System.currentTimeMillis()));
@@ -1061,9 +1168,6 @@ public class TestInstanceOperation extends ZkTestBase {
// Store original EV
Map<String, ExternalView> originalEVs = getEVs();
-
- Map<String, String> swapOutInstancesToSwapInInstances = new HashMap<>();
-
// Get the SWAP_OUT instance.
String instanceToSwapOutName = _participants.get(0).getInstanceName();
InstanceConfig instanceToSwapOutInstanceConfig = _gSetupTool.getClusterManagementTool()
@@ -1071,56 +1175,8 @@ public class TestInstanceOperation extends ZkTestBase {
// Add instance with InstanceOperation set to SWAP_IN
String instanceToSwapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
- swapOutInstancesToSwapInInstances.put(instanceToSwapOutName, instanceToSwapInName);
addParticipant(instanceToSwapInName, instanceToSwapOutInstanceConfig.getLogicalId(LOGICAL_ID),
instanceToSwapOutInstanceConfig.getDomainAsMap().get(ZONE), null, false, -1);
-
- // Validate that the assignment has not changed since setting the InstanceOperation to SWAP_OUT
- Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
- validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
- Collections.emptySet(), Collections.emptySet());
-
- // After the SWAP_IN instance is added, we set the InstanceOperation to SWAP_OUT
- _gSetupTool.getClusterManagementTool().setInstanceOperation(CLUSTER_NAME, instanceToSwapOutName,
- InstanceConstants.InstanceOperation.SWAP_OUT);
-
- Assert.assertTrue(_clusterVerifier.verifyByPolling());
-
- // Enable the SWAP_IN instance to begin the swap operation.
- _gSetupTool.getClusterManagementTool().enableInstance(CLUSTER_NAME, instanceToSwapInName, true);
-
- // Validate that partitions on SWAP_OUT instance does not change after setting the InstanceOperation to SWAP_OUT
- // and adding the SWAP_IN instance to the cluster.
- // Check that the SWAP_IN instance has the same partitions as the SWAP_OUT instance
- // but none of them are in a top state.
- Assert.assertTrue(_bestPossibleClusterVerifier.verifyByPolling());
- validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
- Set.of(instanceToSwapInName), Collections.emptySet());
-
- // Validate that the SWAP_OUT instance is in routing tables and SWAP_IN is not.
- validateRoutingTablesInstance(getEVs(), instanceToSwapOutName, true);
- validateRoutingTablesInstance(getEVs(), instanceToSwapInName, false);
-
- // Assert canSwapBeCompleted is true
- Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .canCompleteSwap(CLUSTER_NAME, instanceToSwapOutName));
- // Assert completeSwapIfPossible is true
- Assert.assertTrue(_gSetupTool.getClusterManagementTool()
- .completeSwapIfPossible(CLUSTER_NAME, instanceToSwapOutName));
-
- Assert.assertTrue(_clusterVerifier.verifyByPolling());
-
- // Validate that the SWAP_IN instance is now in the routing tables.
- validateRoutingTablesInstance(getEVs(), instanceToSwapInName, true);
-
- // Assert that SWAP_OUT instance is disabled and has no partitions assigned to it.
- Assert.assertFalse(_gSetupTool.getClusterManagementTool()
- .getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName).getInstanceEnabled());
-
- // Validate that the SWAP_IN instance has the same partitions the SWAP_OUT instance had before
- // swap was completed.
- verifier(() -> (validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
- Collections.emptySet(), Set.of(instanceToSwapInName))), TIMEOUT);
}
@Test(dependsOnMethods = "testNodeSwapAddSwapInFirst")
@@ -1322,6 +1378,29 @@ public class TestInstanceOperation extends ZkTestBase {
dropTestDBs(ImmutableSet.of("TEST_DB3_DELAYED_CRUSHED", "TEST_DB4_DELAYED_WAGED"));
}
+ @Test(expectedExceptions = HelixException.class, dependsOnMethods = "testEvacuationWithOfflineInstancesInCluster")
+ public void testSwapEvacuateAddRemoveEvacuate() {
+ System.out.println("START TestInstanceOperation.testSwapEvacuateAddRemoveEvacuate() at " + new Date(
+ System.currentTimeMillis()));
+ removeOfflineOrDisabledOrSwapInInstances();
+
+ // Set instance's InstanceOperation to EVACUATE
+ String instanceToSwapOutName = _participants.get(0).getInstanceName();
+ InstanceConfig instanceToSwapOutInstanceConfig = _gSetupTool.getClusterManagementTool()
+ .getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName);
+ _gSetupTool.getClusterManagementTool().setInstanceOperation(CLUSTER_NAME, instanceToSwapOutName,
+ InstanceConstants.InstanceOperation.EVACUATE);
+
+ // Add instance with InstanceOperation set to SWAP_IN
+ String instanceToSwapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
+ addParticipant(instanceToSwapInName, instanceToSwapOutInstanceConfig.getLogicalId(LOGICAL_ID),
+ instanceToSwapOutInstanceConfig.getDomainAsMap().get(ZONE), null, true, -1);
+
+ // Remove EVACUATE instance's InstanceOperation
+ _gSetupTool.getClusterManagementTool()
+ .setInstanceOperation(CLUSTER_NAME, instanceToSwapOutName, null);
+ }
+
/**
* Verifies that the given verifier returns true within the given timeout. Handles AssertionError
* by returning false, which TestHelper.verify will not do. Asserts that return value from
diff --git a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
index aa93bc9c8..60ee9b625 100644
--- a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
@@ -24,12 +24,9 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import org.apache.helix.AccessOption;
import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixException;
-import org.apache.helix.HelixManager;
import org.apache.helix.PropertyPathBuilder;
import org.apache.helix.PropertyType;
import org.apache.helix.api.status.ClusterManagementMode;
@@ -50,8 +47,7 @@ import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.helix.zookeeper.zkclient.DataUpdater;
-
+import org.apache.helix.HelixManager;
public class MockHelixAdmin implements HelixAdmin {
@@ -562,7 +558,7 @@ public class MockHelixAdmin implements HelixAdmin {
}
@Override
- public boolean completeSwapIfPossible(String clusterName, String instanceName) {
+ public boolean completeSwapIfPossible(String clusterName, String instanceName, boolean forceComplete) {
return false;
}
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/PerInstanceAccessor.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/PerInstanceAccessor.java
index f380975a3..bf0e74653 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/PerInstanceAccessor.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/PerInstanceAccessor.java
@@ -389,7 +389,8 @@ public class PerInstanceAccessor extends AbstractHelixResource {
@PathParam("instanceName") String instanceName, @QueryParam("command") String command,
@QueryParam("instanceOperation") InstanceConstants.InstanceOperation state,
@QueryParam("instanceDisabledType") String disabledType,
- @QueryParam("instanceDisabledReason") String disabledReason, String content) {
+ @QueryParam("instanceDisabledReason") String disabledReason,
+ @QueryParam("force") boolean force, String content) {
Command cmd;
try {
cmd = Command.valueOf(command);
@@ -440,7 +441,7 @@ public class PerInstanceAccessor extends AbstractHelixResource {
Map.of("successful", admin.canCompleteSwap(clusterId, instanceName))));
case completeSwapIfPossible:
return OK(OBJECT_MAPPER.writeValueAsString(
- Map.of("successful", admin.completeSwapIfPossible(clusterId, instanceName))));
+ Map.of("successful", admin.completeSwapIfPossible(clusterId, instanceName, force))));
case addInstanceTag:
if (!validInstance(node, instanceName)) {
return badRequest("Instance names are not match!");