// check the instance is alive
LiveInstance liveInstance =
accessor.getProperty(keyBuilder.liveInstance(instanceName));
if (liveInstance == null)
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because " + instanceName
+ " is not alive");
}
// check resource group exists
IdealState idealState = accessor.getProperty(keyBuilder.idealStates(resourceName));
if (idealState == null)
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because " + resourceName
+ " is not added");
}
// check partition exists in resource group
Set<String> resetPartitionNames = new HashSet<String>(partitionNames);
if (idealState.getIdealStateMode() == IdealStateModeProperty.CUSTOMIZED)
{
Set<String> partitions =
new HashSet<String>(idealState.getRecord().getMapFields().keySet());
if (!partitions.containsAll(resetPartitionNames))
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because not all "
+ partitionNames + " exist");
}
}
else
{
Set<String> partitions =
new HashSet<String>(idealState.getRecord().getListFields().keySet());
if (!partitions.containsAll(resetPartitionNames))
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because not all "
+ partitionNames + " exist");
}
}
// check partition is in ERROR state
String sessionId = liveInstance.getSessionId();
CurrentState curState =
accessor.getProperty(keyBuilder.currentState(instanceName,
sessionId,
resourceName));
for (String partitionName : resetPartitionNames)
{
if (!curState.getState(partitionName).equals("ERROR"))
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because not all "
+ partitionNames + " are in ERROR state");
}
}
// check stateModelDef exists and get initial state
String stateModelDef = idealState.getStateModelDefRef();
StateModelDefinition stateModel =
accessor.getProperty(keyBuilder.stateModelDef(stateModelDef));
if (stateModel == null)
{
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName + ", because " + stateModelDef
+ " is NOT found");
}
// check there is no pending messages for the partitions exist
List<Message> messages = accessor.getChildValues(keyBuilder.messages(instanceName));
for (Message message : messages)
{
if (!MessageType.STATE_TRANSITION.toString().equalsIgnoreCase(message.getMsgType())
|| !sessionId.equals(message.getTgtSessionId())
|| !resourceName.equals(message.getResourceName())
|| !resetPartitionNames.contains(message.getPartitionName()))
{
continue;
}
throw new HelixException("Can't reset state for " + resourceName + "/"
+ partitionNames + " on " + instanceName
+ ", because a pending message exists: " + message);
}
String adminName = null;