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 <[email protected]>
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!");
