You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by zz...@apache.org on 2013/08/29 08:17:47 UTC
[3/4] [HELIX-109] Review Helix model package,
merge model changes with accessor changes, incomplete, rb=13878
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
----------------------------------------------------------------------
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 9564e35..6d64bb8 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
@@ -71,7 +71,7 @@ public class AutoRebalancer implements Rebalancer {
public ResourceAssignment computeResourceMapping(Resource resource, IdealState currentIdealState,
CurrentStateOutput currentStateOutput, ClusterDataCache clusterData) {
// Compute a preference list based on the current ideal state
- List<String> partitions = new ArrayList<String>(currentIdealState.getPartitionSet());
+ List<String> partitions = new ArrayList<String>(currentIdealState.getPartitionStringSet());
String stateModelName = currentIdealState.getStateModelDefRef();
StateModelDefinition stateModelDef = clusterData.getStateModelDef(stateModelName);
Map<String, LiveInstance> liveInstance = clusterData.getLiveInstances();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
index 8557fa0..5de420c 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
@@ -102,7 +102,7 @@ public class CustomRebalancer implements Rebalancer {
HelixDefinedState.ERROR.toString()))
&& disabledInstancesForPartition.contains(instance)) {
// if disabled and not in ERROR state, transit to initial-state (e.g. OFFLINE)
- instanceStateMap.put(instance, stateModelDef.getInitialState());
+ instanceStateMap.put(instance, stateModelDef.getInitialStateString());
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/ConstraintBasedAssignment.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/ConstraintBasedAssignment.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/ConstraintBasedAssignment.java
index d2dbdef..ee7524f 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/ConstraintBasedAssignment.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/util/ConstraintBasedAssignment.java
@@ -28,8 +28,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.helix.HelixDefinedState;
import org.apache.helix.HelixConstants.StateModelToken;
+import org.apache.helix.HelixDefinedState;
import org.apache.helix.controller.stages.ClusterDataCache;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.LiveInstance;
@@ -86,7 +86,7 @@ public class ConstraintBasedAssignment {
HelixDefinedState.ERROR.toString()))
&& disabledInstancesForPartition.contains(instance)) {
// if disabled and not in ERROR state, transit to initial-state (e.g. OFFLINE)
- instanceStateMap.put(instance, stateModelDef.getInitialState());
+ instanceStateMap.put(instance, stateModelDef.getInitialStateString());
}
}
}
@@ -96,7 +96,7 @@ public class ConstraintBasedAssignment {
return instanceStateMap;
}
- List<String> statesPriorityList = stateModelDef.getStatesPriorityList();
+ List<String> statesPriorityList = stateModelDef.getStatesPriorityStringList();
boolean assigned[] = new boolean[instancePreferenceList.size()];
Map<String, LiveInstance> liveInstancesMap = cache.getLiveInstances();
@@ -151,7 +151,7 @@ public class ConstraintBasedAssignment {
public static LinkedHashMap<String, Integer> stateCount(StateModelDefinition stateModelDef,
int liveNodesNb, int totalReplicas) {
LinkedHashMap<String, Integer> stateCountMap = new LinkedHashMap<String, Integer>();
- List<String> statesPriorityList = stateModelDef.getStatesPriorityList();
+ List<String> statesPriorityList = stateModelDef.getStatesPriorityStringList();
int replicas = totalReplicas;
for (String state : statesPriorityList) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
index b90880e..062f468 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
@@ -72,7 +72,7 @@ public class ClusterDataCache {
_liveInstanceMap = accessor.getChildValuesMap(keyBuilder.liveInstances());
for (LiveInstance instance : _liveInstanceMap.values()) {
- LOG.trace("live instance: " + instance.getInstanceName() + " " + instance.getSessionId());
+ LOG.trace("live instance: " + instance.getInstanceName() + " " + instance.getSessionIdString());
}
_stateModelDefMap = accessor.getChildValuesMap(keyBuilder.stateModelDefs());
@@ -90,7 +90,7 @@ public class ClusterDataCache {
new HashMap<String, Map<String, Map<String, CurrentState>>>();
for (String instanceName : _liveInstanceMap.keySet()) {
LiveInstance liveInstance = _liveInstanceMap.get(instanceName);
- String sessionId = liveInstance.getSessionId();
+ String sessionId = liveInstance.getSessionIdString();
if (!allCurStateMap.containsKey(instanceName)) {
allCurStateMap.put(instanceName, new HashMap<String, Map<String, CurrentState>>());
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
index d8f98ed..a1b7b0d 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
@@ -46,7 +46,7 @@ public class CompatibilityCheckStage extends AbstractBaseStage {
HelixManagerProperties properties = manager.getProperties();
Map<String, LiveInstance> liveInstanceMap = cache.getLiveInstances();
for (LiveInstance liveInstance : liveInstanceMap.values()) {
- String participantVersion = liveInstance.getHelixVersion();
+ String participantVersion = liveInstance.getHelixVersionString();
if (!properties.isParticipantCompatible(participantVersion)) {
String errorMsg =
"incompatible participant. pipeline will not continue. " + "controller: "
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
index 6097432..1e4e72b 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
@@ -57,7 +57,7 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
if (!MessageType.STATE_TRANSITION.toString().equalsIgnoreCase(message.getMsgType())) {
continue;
}
- if (!instance.getSessionId().equals(message.getTgtSessionId())) {
+ if (!instance.getSessionIdString().equals(message.getTgtSessionIdString())) {
continue;
}
String resourceName = message.getResourceName();
@@ -71,7 +71,7 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
Partition partition = resource.getPartition(partitionName);
if (partition != null) {
currentStateOutput.setPendingState(resourceName, partition, instanceName,
- message.getToState());
+ message.getToStateString());
} else {
// log
}
@@ -82,7 +82,7 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
Partition partition = resource.getPartition(partitionName);
if (partition != null) {
currentStateOutput.setPendingState(resourceName, partition, instanceName,
- message.getToState());
+ message.getToStateString());
} else {
// log
}
@@ -94,12 +94,12 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
for (LiveInstance instance : liveInstances.values()) {
String instanceName = instance.getInstanceName();
- String clientSessionId = instance.getSessionId();
+ String clientSessionId = instance.getSessionIdString();
Map<String, CurrentState> currentStateMap =
cache.getCurrentState(instanceName, clientSessionId);
for (CurrentState currentState : currentStateMap.values()) {
- if (!instance.getSessionId().equals(currentState.getSessionId())) {
+ if (!instance.getSessionIdString().equals(currentState.getSessionIdString())) {
continue;
}
String resourceName = currentState.getResourceName();
@@ -114,7 +114,7 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
currentStateOutput.setBucketSize(resourceName, currentState.getBucketSize());
- Map<String, String> partitionStateMap = currentState.getPartitionStateMap();
+ Map<String, String> partitionStateMap = currentState.getPartitionStateStringMap();
for (String partitionName : partitionStateMap.keySet()) {
Partition partition = resource.getPartition(partitionName);
if (partition != null) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
index 35ef177..d50bd9e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ExternalViewComputeStage.java
@@ -168,7 +168,7 @@ public class ExternalViewComputeStage extends AbstractBaseStage {
Builder keyBuilder = accessor.keyBuilder();
- for (String taskPartitionName : ev.getPartitionSet()) {
+ for (String taskPartitionName : ev.getPartitionStringSet()) {
for (String taskState : ev.getStateMap(taskPartitionName).values()) {
if (taskState.equalsIgnoreCase(HelixDefinedState.ERROR.toString())
|| taskState.equalsIgnoreCase("COMPLETED")) {
@@ -193,7 +193,7 @@ public class ExternalViewComputeStage extends AbstractBaseStage {
}
}
// fill the controllerMsgIdCountMap
- for (String taskId : taskQueueIdealState.getPartitionSet()) {
+ for (String taskId : taskQueueIdealState.getPartitionStringSet()) {
String controllerMsgId =
taskQueueIdealState.getRecord().getMapField(taskId)
.get(DefaultSchedulerMessageHandlerFactory.CONTROLLER_MSG_ID);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
index 92964e9..d0237d7 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
@@ -64,7 +64,7 @@ public class MessageGenerationPhase extends AbstractBaseStage {
Map<String, String> sessionIdMap = new HashMap<String, String>();
for (LiveInstance liveInstance : liveInstances.values()) {
- sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionId());
+ sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionIdString());
}
MessageGenerationOutput output = new MessageGenerationOutput();
@@ -89,7 +89,7 @@ public class MessageGenerationPhase extends AbstractBaseStage {
String currentState =
currentStateOutput.getCurrentState(resourceName, partition, instanceName);
if (currentState == null) {
- currentState = stateModelDef.getInitialState();
+ currentState = stateModelDef.getInitialStateString();
}
if (desiredState.equalsIgnoreCase(currentState)) {
@@ -170,7 +170,7 @@ public class MessageGenerationPhase extends AbstractBaseStage {
}
// add generated messages to output according to state priority
- List<String> statesPriorityList = stateModelDef.getStatesPriorityList();
+ List<String> statesPriorityList = stateModelDef.getStatesPriorityStringList();
for (String state : statesPriorityList) {
if (messageMap.containsKey(state)) {
for (Message message : messageMap.get(state)) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
index 9a420aa..8478db3 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
@@ -104,7 +104,7 @@ public class MessageSelectionStage extends AbstractBaseStage {
selectMessages(cache.getLiveInstances(),
currentStateOutput.getCurrentStateMap(resourceName, partition),
currentStateOutput.getPendingStateMap(resourceName, partition), messages,
- stateConstraints, stateTransitionPriorities, stateModelDef.getInitialState());
+ stateConstraints, stateTransitionPriorities, stateModelDef.getInitialStateString());
output.addMessages(resourceName, partition, selectedMessages);
}
}
@@ -169,8 +169,8 @@ public class MessageSelectionStage extends AbstractBaseStage {
Map<Integer, List<Message>> messagesGroupByStateTransitPriority =
new TreeMap<Integer, List<Message>>();
for (Message message : messages) {
- String fromState = message.getFromState();
- String toState = message.getToState();
+ String fromState = message.getFromStateString();
+ String toState = message.getToStateString();
String transition = fromState + "-" + toState;
int priority = Integer.MAX_VALUE;
@@ -187,8 +187,8 @@ public class MessageSelectionStage extends AbstractBaseStage {
// select messages
for (List<Message> messageList : messagesGroupByStateTransitPriority.values()) {
for (Message message : messageList) {
- String fromState = message.getFromState();
- String toState = message.getToState();
+ String fromState = message.getFromStateString();
+ String toState = message.getToStateString();
if (!bounds.containsKey(fromState)) {
LOG.error("Message's fromState is not in currentState. message: " + message);
@@ -239,7 +239,7 @@ public class MessageSelectionStage extends AbstractBaseStage {
IdealState idealState, ClusterDataCache cache) {
Map<String, Bounds> stateConstraints = new HashMap<String, Bounds>();
- List<String> statePriorityList = stateModelDefinition.getStatesPriorityList();
+ List<String> statePriorityList = stateModelDefinition.getStatesPriorityStringList();
for (String state : statePriorityList) {
String numInstancesPerState = stateModelDefinition.getNumInstancesPerState(state);
int max = -1;
@@ -272,7 +272,7 @@ public class MessageSelectionStage extends AbstractBaseStage {
// so that behavior is consistent
private Map<String, Integer> getStateTransitionPriorityMap(StateModelDefinition stateModelDef) {
Map<String, Integer> stateTransitionPriorities = new HashMap<String, Integer>();
- List<String> stateTransitionPriorityList = stateModelDef.getStateTransitionPriorityList();
+ List<String> stateTransitionPriorityList = stateModelDef.getStateTransitionPriorityStringList();
for (int i = 0; i < stateTransitionPriorityList.size(); i++) {
stateTransitionPriorities.put(stateTransitionPriorityList.get(i), i);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/RebalanceIdealStateStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/RebalanceIdealStateStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/RebalanceIdealStateStage.java
index d82ee2f..f16bb39 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/RebalanceIdealStateStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/RebalanceIdealStateStage.java
@@ -60,7 +60,7 @@ public class RebalanceIdealStateStage extends AbstractBaseStage {
(Rebalancer) (HelixUtil.loadClass(getClass(), rebalancerClassName).newInstance());
balancer.init(manager);
Resource resource = new Resource(resourceName);
- for (String partitionName : currentIdealState.getPartitionSet()) {
+ for (String partitionName : currentIdealState.getPartitionStringSet()) {
resource.addPartition(partitionName);
}
ResourceAssignment resourceAssignment =
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
index 51f0ec1..b67a426 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
@@ -53,7 +53,7 @@ public class ResourceComputationStage extends AbstractBaseStage {
if (idealStates != null && idealStates.size() > 0) {
for (IdealState idealState : idealStates.values()) {
- Set<String> partitionSet = idealState.getPartitionSet();
+ Set<String> partitionSet = idealState.getPartitionStringSet();
String resourceName = idealState.getResourceName();
for (String partition : partitionSet) {
@@ -74,7 +74,7 @@ public class ResourceComputationStage extends AbstractBaseStage {
if (availableInstances != null && availableInstances.size() > 0) {
for (LiveInstance instance : availableInstances.values()) {
String instanceName = instance.getInstanceName();
- String clientSessionId = instance.getSessionId();
+ String clientSessionId = instance.getSessionIdString();
Map<String, CurrentState> currentStateMap =
cache.getCurrentState(instanceName, clientSessionId);
@@ -84,7 +84,7 @@ public class ResourceComputationStage extends AbstractBaseStage {
for (CurrentState currentState : currentStateMap.values()) {
String resourceName = currentState.getResourceName();
- Map<String, String> resourceStateMap = currentState.getPartitionStateMap();
+ Map<String, String> resourceStateMap = currentState.getPartitionStateStringMap();
// don't overwrite ideal state settings
if (!resourceMap.containsKey(resourceName)) {
@@ -98,8 +98,8 @@ public class ResourceComputationStage extends AbstractBaseStage {
if (currentState.getStateModelDefRef() == null) {
LOG.error("state model def is null." + "resource:" + currentState.getResourceName()
- + ", partitions: " + currentState.getPartitionStateMap().keySet() + ", states: "
- + currentState.getPartitionStateMap().values());
+ + ", partitions: " + currentState.getPartitionStateStringMap().keySet() + ", states: "
+ + currentState.getPartitionStateStringMap().values());
throw new StageException("State model def is null for resource:"
+ currentState.getResourceName());
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
index 192a645..85ae163 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
@@ -96,7 +96,7 @@ public class TaskAssignmentStage extends AbstractBaseStage {
LiveInstance liveInstance = liveInstanceMap.get(instanceName);
String participantVersion = null;
if (liveInstance != null) {
- participantVersion = liveInstance.getHelixVersion();
+ participantVersion = liveInstance.getHelixVersionString();
}
if (resource == null || !resource.getBatchMessageMode() || participantVersion == null
@@ -106,9 +106,9 @@ public class TaskAssignmentStage extends AbstractBaseStage {
}
String key =
- keyBuilder.currentState(message.getTgtName(), message.getTgtSessionId(),
+ keyBuilder.currentState(message.getTgtName(), message.getTgtSessionIdString(),
message.getResourceName()).getPath()
- + "/" + message.getFromState() + "/" + message.getToState();
+ + "/" + message.getFromStateString() + "/" + message.getToStateString();
if (!batchMessages.containsKey(key)) {
Message batchMessage = new Message(message.getRecord());
@@ -131,9 +131,9 @@ public class TaskAssignmentStage extends AbstractBaseStage {
List<PropertyKey> keys = new ArrayList<PropertyKey>();
for (Message message : messages) {
- logger.info("Sending Message " + message.getMsgId() + " to " + message.getTgtName()
+ logger.info("Sending Message " + message.getMsgIdString() + " to " + message.getTgtName()
+ " transit " + message.getPartitionName() + "|" + message.getPartitionNames() + " from:"
- + message.getFromState() + " to:" + message.getToState());
+ + message.getFromStateString() + " to:" + message.getToStateString());
// System.out.println("[dbg] Sending Message " + message.getMsgId() + " to " +
// message.getTgtName()
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
index 1ed6dea..0fb6709 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
@@ -150,7 +150,7 @@ public class ControllerManager extends AbstractManager {
LiveInstance leader = _dataAccessor.getProperty(_keyBuilder.controllerLeader());
if (leader != null) {
String leaderName = leader.getInstanceName();
- String sessionId = leader.getSessionId();
+ String sessionId = leader.getSessionIdString();
if (leaderName != null && leaderName.equals(_instanceName) && sessionId != null
&& sessionId.equals(_sessionId)) {
return true;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
index b96de18..3fd3fb8 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
@@ -57,7 +57,7 @@ class CurStateCarryOverUpdater implements DataUpdater<ZNRecord> {
curState = new CurrentState(currentData);
}
- for (String partitionName : _lastCurState.getPartitionStateMap().keySet()) {
+ for (String partitionName : _lastCurState.getPartitionStateStringMap().keySet()) {
// carry-over only when current-state not exist
if (curState.getState(partitionName) == null) {
curState.setState(partitionName, _initState);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
index 5f6d083..523adba 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
@@ -36,7 +36,7 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ message.getMsgType());
}
@@ -63,18 +63,18 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
String type = _message.getMsgType();
HelixTaskResult result = new HelixTaskResult();
if (!type.equals(MessageType.CONTROLLER_MSG.toString())) {
- throw new HelixException("Unexpected msg type for message " + _message.getMsgId()
+ throw new HelixException("Unexpected msg type for message " + _message.getMsgIdString()
+ " type:" + _message.getMsgType());
}
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
+ "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
result.setSuccess(true);
return result;
}
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgIdString(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
index d2e56eb..a147ddb 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
@@ -99,7 +99,7 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgIdString(), e);
}
}
@@ -109,7 +109,7 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ message.getMsgType());
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
index 5451a81..a7fcc10 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
@@ -76,7 +76,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
@Override
public void onTimeOut() {
- _logger.info("Scheduler msg timeout " + _originalMessage.getMsgId() + " timout with "
+ _logger.info("Scheduler msg timeout " + _originalMessage.getMsgIdString() + " timout with "
+ _timeout + " Ms");
_statusUpdateUtil.logError(_originalMessage, SchedulerAsyncCallback.class, "Task timeout",
@@ -86,13 +86,13 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
@Override
public void onReplyMessage(Message message) {
- _logger.info("Update for scheduler msg " + _originalMessage.getMsgId() + " Message "
+ _logger.info("Update for scheduler msg " + _originalMessage.getMsgIdString() + " Message "
+ message.getMsgSrc() + " id " + message.getCorrelationId() + " completed");
String key = "MessageResult " + message.getMsgSrc() + " " + UUID.randomUUID();
_resultSummaryMap.put(key, message.getResultMap());
if (this.isDone()) {
- _logger.info("Scheduler msg " + _originalMessage.getMsgId() + " completed");
+ _logger.info("Scheduler msg " + _originalMessage.getMsgIdString() + " completed");
_statusUpdateUtil.logInfo(_originalMessage, SchedulerAsyncCallback.class,
"Scheduler task completed", _manager.getHelixDataAccessor());
addSummary(_resultSummaryMap, _originalMessage, _manager, false);
@@ -111,12 +111,12 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
ZNRecord statusUpdate =
accessor.getProperty(
keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- originalMessage.getMsgId())).getRecord();
+ originalMessage.getMsgIdString())).getRecord();
statusUpdate.getMapFields().putAll(_resultSummaryMap);
accessor.setProperty(
keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- originalMessage.getMsgId()), new StatusUpdate(statusUpdate));
+ originalMessage.getMsgIdString()), new StatusUpdate(statusUpdate));
}
}
@@ -133,7 +133,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ message.getMsgType());
}
@@ -206,7 +206,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
newAddedScheduledTasks.getRecord().setMapField(partitionId,
task.getRecord().getSimpleFields());
_logger.info("Scheduling for controllerMsg " + controllerMsgId + " , sending task "
- + partitionId + " " + task.getMsgId() + " to " + instanceName);
+ + partitionId + " " + task.getMsgIdString() + " to " + instanceName);
if (_logger.isDebugEnabled()) {
_logger.debug(task.getRecord().getSimpleFields());
@@ -223,16 +223,16 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
ZNRecord statusUpdate =
accessor.getProperty(
keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgId())).getRecord();
+ _message.getMsgIdString())).getRecord();
statusUpdate.getMapFields().put("SentMessageCount", sendSummary);
accessor.updateProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgId()), new StatusUpdate(statusUpdate));
+ _message.getMsgIdString()), new StatusUpdate(statusUpdate));
}
private int findTopPartitionId(IdealState currentTaskQueue) {
int topId = 0;
- for (String partitionName : currentTaskQueue.getPartitionSet()) {
+ for (String partitionName : currentTaskQueue.getPartitionStringSet()) {
try {
String partitionNumStr = partitionName.substring(partitionName.lastIndexOf('_') + 1);
int num = Integer.parseInt(partitionNumStr);
@@ -251,7 +251,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
String type = _message.getMsgType();
HelixTaskResult result = new HelixTaskResult();
if (!type.equals(MessageType.SCHEDULER_MSG.toString())) {
- throw new HelixException("Unexpected msg type for message " + _message.getMsgId()
+ throw new HelixException("Unexpected msg type for message " + _message.getMsgIdString()
+ " type:" + _message.getMsgType());
}
// Parse timeout value
@@ -298,11 +298,11 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
if (InstanceType.PARTICIPANT == recipientCriteria.getRecipientInstanceType()
&& hasSchedulerTaskQueue) {
handleMessageUsingScheduledTaskQueue(recipientCriteria, messageTemplate,
- _message.getMsgId());
+ _message.getMsgIdString());
result.setSuccess(true);
- result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId());
+ result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgIdString());
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
+ "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
return result;
}
@@ -328,23 +328,23 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
ZNRecord statusUpdate =
accessor.getProperty(
keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgId())).getRecord();
+ _message.getMsgIdString())).getRecord();
statusUpdate.getMapFields().put("SentMessageCount", sendSummary);
accessor.setProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgId()), new StatusUpdate(statusUpdate));
+ _message.getMsgIdString()), new StatusUpdate(statusUpdate));
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
- result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId());
+ "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
+ result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgIdString());
result.setSuccess(true);
return result;
}
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgIdString(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
index c9ad0f3..71a5c4a 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
@@ -175,7 +175,7 @@ public class DistributedControllerManager extends AbstractManager {
LiveInstance leader = _dataAccessor.getProperty(_keyBuilder.controllerLeader());
if (leader != null) {
String leaderName = leader.getInstanceName();
- String sessionId = leader.getSessionId();
+ String sessionId = leader.getSessionIdString();
if (leaderName != null && leaderName.equals(_instanceName) && sessionId != null
&& sessionId.equals(_sessionId)) {
return true;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
index 0ab8342..af33b03 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
@@ -138,7 +138,7 @@ public class DistributedLeaderElection implements ControllerChangeListener {
leader = accessor.getProperty(keyBuilder.controllerLeader());
if (leader != null) {
- String leaderSessionId = leader.getSessionId();
+ String leaderSessionId = leader.getSessionIdString();
LOG.info("Leader exists for cluster: " + manager.getClusterName() + ", currentLeader: "
+ leader.getInstanceName() + ", leaderSessionId: " + leaderSessionId);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
index 70dd592..ff932a9 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
@@ -151,14 +151,14 @@ public class ParticipantManagerHelper {
* update sessionId field in live-instance if necessary
*/
LiveInstance curLiveInstance = new LiveInstance(record);
- if (!curLiveInstance.getSessionId().equals(_sessionId)) {
+ if (!curLiveInstance.getSessionIdString().equals(_sessionId)) {
/**
* in last handle-new-session,
* live-instance is created by new zkconnection with stale session-id inside
* just update session-id field
*/
LOG.info("overwriting session-id by ephemeralOwner: " + ephemeralOwner
- + ", old-sessionId: " + curLiveInstance.getSessionId() + ", new-sessionId: "
+ + ", old-sessionId: " + curLiveInstance.getSessionIdString() + ", new-sessionId: "
+ _sessionId);
curLiveInstance.setSessionId(_sessionId);
@@ -231,7 +231,7 @@ public class ParticipantManagerHelper {
_keyBuilder.currentState(_instanceName, _sessionId, lastCurState.getResourceName())
.getPath();
_dataAccessor.getBaseDataAccessor().update(curStatePath,
- new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialState(), lastCurState),
+ new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialStateString(), lastCurState),
AccessOption.PERSISTENT);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
----------------------------------------------------------------------
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 754df7b..8172c61 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
@@ -326,7 +326,7 @@ public class ZKHelixAdmin implements HelixAdmin {
}
// check partition is in ERROR state
- String sessionId = liveInstance.getSessionId();
+ String sessionId = liveInstance.getSessionIdString();
CurrentState curState =
accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, resourceName));
for (String partitionName : resetPartitionNames) {
@@ -348,7 +348,7 @@ public class ZKHelixAdmin implements HelixAdmin {
List<Message> messages = accessor.getChildValues(keyBuilder.messages(instanceName));
for (Message message : messages) {
if (!MessageType.STATE_TRANSITION.toString().equalsIgnoreCase(message.getMsgType())
- || !sessionId.equals(message.getTgtSessionId())
+ || !sessionId.equals(message.getTgtSessionIdString())
|| !resourceName.equals(message.getResourceName())
|| !resetPartitionNames.contains(message.getPartitionName())) {
continue;
@@ -381,7 +381,7 @@ public class ZKHelixAdmin implements HelixAdmin {
message.setTgtSessionId(sessionId);
message.setStateModelDef(stateModelDef);
message.setFromState(HelixDefinedState.ERROR.toString());
- message.setToState(stateModel.getInitialState());
+ message.setToState(stateModel.getInitialStateString());
message.setStateModelFactoryName(idealState.getStateModelFactoryName());
resetMessages.add(message);
@@ -986,7 +986,7 @@ public class ZKHelixAdmin implements HelixAdmin {
}
// StateModelDefinition def = new StateModelDefinition(stateModDef);
- List<String> statePriorityList = stateModDef.getStatesPriorityList();
+ List<String> statePriorityList = stateModDef.getStatesPriorityStringList();
String masterStateValue = null;
String slaveStateValue = null;
@@ -1144,7 +1144,7 @@ public class ZKHelixAdmin implements HelixAdmin {
@Override
public void rebalance(String clusterName, IdealState currentIdealState, List<String> instanceNames) {
Set<String> activeInstances = new HashSet<String>();
- for (String partition : currentIdealState.getPartitionSet()) {
+ for (String partition : currentIdealState.getPartitionStringSet()) {
activeInstances.addAll(currentIdealState.getRecord().getListField(partition));
}
instanceNames.removeAll(activeInstances);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
index 025402d..087d2fb 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
@@ -76,7 +76,7 @@ public class ZKHelixDataAccessor implements HelixDataAccessor, ControllerChangeL
PropertyType type = key.getType();
String path = key.getPath();
int options = constructOptions(type);
- return _baseDataAccessor.create(path, value.getRecord(), options);
+ return _baseDataAccessor.create(path, value == null ? null : value.getRecord(), options);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index 621c18b..715d683 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -839,7 +839,7 @@ public class ZKHelixManager implements HelixManager {
keyBuilder.currentState(_instanceName, _sessionId, lastCurState.getResourceName())
.getPath();
_helixAccessor.getBaseDataAccessor().update(curStatePath,
- new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialState(), lastCurState),
+ new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialStateString(), lastCurState),
AccessOption.PERSISTENT);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/AsyncCallback.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/AsyncCallback.java b/helix-core/src/main/java/org/apache/helix/messaging/AsyncCallback.java
index f9743a4..0f031dc 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/AsyncCallback.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/AsyncCallback.java
@@ -71,7 +71,7 @@ public abstract class AsyncCallback {
}
public synchronized final void onReply(Message message) {
- _logger.info("OnReply msg " + message.getMsgId());
+ _logger.info("OnReply msg " + message.getMsgIdString());
if (!isDone()) {
_messageReplied.add(message);
try {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java b/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
index 2eec354..c22abb8 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
@@ -157,7 +157,7 @@ public class DefaultMessagingService implements ClusterMessagingService {
List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances());
for (LiveInstance liveInstance : liveInstances) {
- sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionId());
+ sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionIdString());
}
}
for (Map<String, String> map : matchedList) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
index c218a15..8375420 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
@@ -56,14 +56,14 @@ public class AsyncCallbackService implements MessageHandlerFactory {
void verifyMessage(Message message) {
if (!message.getMsgType().toString().equalsIgnoreCase(MessageType.TASK_REPLY.toString())) {
String errorMsg =
- "Unexpected msg type for message " + message.getMsgId() + " type:" + message.getMsgType()
+ "Unexpected msg type for message " + message.getMsgIdString() + " type:" + message.getMsgType()
+ " Expected : " + MessageType.TASK_REPLY;
_logger.error(errorMsg);
throw new HelixException(errorMsg);
}
String correlationId = message.getCorrelationId();
if (correlationId == null) {
- String errorMsg = "Message " + message.getMsgId() + " does not have correlation id";
+ String errorMsg = "Message " + message.getMsgIdString() + " does not have correlation id";
_logger.error(errorMsg);
throw new HelixException(errorMsg);
}
@@ -71,13 +71,13 @@ public class AsyncCallbackService implements MessageHandlerFactory {
if (!_callbackMap.containsKey(correlationId)) {
String errorMsg =
"Message "
- + message.getMsgId()
+ + message.getMsgIdString()
+ " does not have correponding callback. Probably timed out already. Correlation id: "
+ correlationId;
_logger.error(errorMsg);
throw new HelixException(errorMsg);
}
- _logger.info("Verified reply message " + message.getMsgId() + " correlation:" + correlationId);
+ _logger.info("Verified reply message " + message.getMsgIdString() + " correlation:" + correlationId);
}
@Override
@@ -110,7 +110,7 @@ public class AsyncCallbackService implements MessageHandlerFactory {
verifyMessage(_message);
HelixTaskResult result = new HelixTaskResult();
assert (_correlationId.equalsIgnoreCase(_message.getCorrelationId()));
- _logger.info("invoking reply message " + _message.getMsgId() + ", correlationid:"
+ _logger.info("invoking reply message " + _message.getMsgIdString() + ", correlationid:"
+ _correlationId);
AsyncCallback callback = _callbackMap.get(_correlationId);
@@ -127,7 +127,7 @@ public class AsyncCallbackService implements MessageHandlerFactory {
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgIdString(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
index 627babc..4c634b9 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
@@ -93,7 +93,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
String partitionName = _message.getPartitionName();
- String fromState = _message.getFromState();
+ String fromState = _message.getFromStateString();
// Verify the fromState and current state of the stateModel
String state = _currentStateDelta.getState(partitionName);
@@ -119,7 +119,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
String partitionKey = _message.getPartitionName();
String resource = _message.getResourceName();
- String sessionId = _message.getTgtSessionId();
+ String sessionId = _message.getTgtSessionIdString();
String instanceName = _manager.getInstanceName();
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
@@ -132,15 +132,15 @@ public class HelixStateTransitionHandler extends MessageHandler {
// new session
// sessionId might change when we update the state model state.
// for zk current state it is OK as we have the per-session current state node
- if (!_message.getTgtSessionId().equals(_manager.getSessionId())) {
+ if (!_message.getTgtSessionIdString().equals(_manager.getSessionId())) {
logger.warn("Session id has changed. Skip postExecutionMessage. Old session "
- + _message.getExecutionSessionId() + " , new session : " + _manager.getSessionId());
+ + _message.getExecutionSessionIdString() + " , new session : " + _manager.getSessionId());
return;
}
if (taskResult.isSuccess()) {
// String fromState = message.getFromState();
- String toState = _message.getToState();
+ String toState = _message.getToStateString();
_currentStateDelta.setState(partitionKey, toState);
if (toState.equalsIgnoreCase(HelixDefinedState.DROPPED.toString())) {
@@ -178,7 +178,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
// state in this case
logger
.error("State transition interrupted but not timeout. Not updating state. Partition : "
- + _message.getPartitionName() + " MsgId : " + _message.getMsgId());
+ + _message.getPartitionName() + " MsgId : " + _message.getMsgIdString());
return;
}
}
@@ -187,7 +187,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
_stateModel.updateState(HelixDefinedState.ERROR.toString());
// if we have errors transit from ERROR state, disable the partition
- if (_message.getFromState().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
+ if (_message.getFromStateString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
disablePartition();
}
}
@@ -226,7 +226,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
HelixAdmin admin = _manager.getClusterManagmentTool();
admin.enablePartition(false, clusterName, instanceName, resourceName,
Arrays.asList(partitionName));
- logger.info("error in transit from ERROR to " + _message.getToState() + " for partition: "
+ logger.info("error in transit from ERROR to " + _message.getToStateString() + " for partition: "
+ partitionName + ". disable it on " + instanceName);
}
@@ -285,8 +285,8 @@ public class HelixStateTransitionHandler extends MessageHandler {
// by default, we invoke state transition function in state model
Method methodToInvoke = null;
- String fromState = message.getFromState();
- String toState = message.getToState();
+ String fromState = message.getFromStateString();
+ String toState = message.getToStateString();
methodToInvoke =
_transitionMethodFinder.getMethodForTransition(_stateModel.getClass(), fromState, toState,
new Class[] {
@@ -332,11 +332,11 @@ public class HelixStateTransitionHandler extends MessageHandler {
_stateModel.updateState(HelixDefinedState.ERROR.toString());
// if transit from ERROR state, disable the partition
- if (_message.getFromState().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
+ if (_message.getFromStateString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
disablePartition();
}
accessor.updateProperty(
- keyBuilder.currentState(instanceName, _message.getTgtSessionId(), resourceName),
+ keyBuilder.currentState(instanceName, _message.getTgtSessionIdString(), resourceName),
currentStateDelta);
}
} finally {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTask.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTask.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTask.java
index d9f7ae2..dd8612e 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTask.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTask.java
@@ -98,14 +98,14 @@ public class HelixTask implements MessageTask {
_statusUpdateUtil.logError(_message, HelixTask.class, e,
"State transition interrupted, timeout:" + _isTimeout, accessor);
- logger.info("Message " + _message.getMsgId() + " is interrupted");
+ logger.info("Message " + _message.getMsgIdString() + " is interrupted");
} catch (Exception e) {
taskResult = new HelixTaskResult();
taskResult.setException(e);
taskResult.setMessage(e.getMessage());
String errorMessage =
- "Exception while executing a message. " + e + " msgId: " + _message.getMsgId()
+ "Exception while executing a message. " + e + " msgId: " + _message.getMsgIdString()
+ " type: " + _message.getMsgType();
logger.error(errorMessage, e);
_statusUpdateUtil.logError(_message, HelixTask.class, e, errorMessage, accessor);
@@ -119,17 +119,17 @@ public class HelixTask implements MessageTask {
if (taskResult.isSuccess()) {
_statusUpdateUtil.logInfo(_message, _handler.getClass(),
"Message handling task completed successfully", accessor);
- logger.info("Message " + _message.getMsgId() + " completed.");
+ logger.info("Message " + _message.getMsgIdString() + " completed.");
} else {
type = ErrorType.INTERNAL;
if (taskResult.isInterrupted()) {
- logger.info("Message " + _message.getMsgId() + " is interrupted");
+ logger.info("Message " + _message.getMsgIdString() + " is interrupted");
code = _isTimeout ? ErrorCode.TIMEOUT : ErrorCode.CANCEL;
if (_isTimeout) {
int retryCount = _message.getRetryCount();
logger.info("Message timeout, retry count: " + retryCount + " msgId:"
- + _message.getMsgId());
+ + _message.getMsgIdString());
_statusUpdateUtil.logInfo(_message, _handler.getClass(),
"Message handling task timeout, retryCount:" + retryCount, accessor);
// Notify the handler that timeout happens, and the number of retries left
@@ -166,12 +166,12 @@ public class HelixTask implements MessageTask {
code = ErrorCode.ERROR;
String errorMessage =
- "Exception after executing a message, msgId: " + _message.getMsgId() + e;
+ "Exception after executing a message, msgId: " + _message.getMsgIdString() + e;
logger.error(errorMessage, e);
_statusUpdateUtil.logError(_message, HelixTask.class, errorMessage, accessor);
} finally {
long end = System.currentTimeMillis();
- logger.info("msg: " + _message.getMsgId() + " handling task completed, results:"
+ logger.info("msg: " + _message.getMsgIdString() + " handling task completed, results:"
+ taskResult.isSuccess() + ", at: " + end + ", took:" + (end - start));
// Notify the handler about any error happened in the handling procedure, so that
@@ -190,9 +190,9 @@ public class HelixTask implements MessageTask {
Builder keyBuilder = accessor.keyBuilder();
if (message.getTgtName().equalsIgnoreCase("controller")) {
// TODO: removeProperty returns boolean
- accessor.removeProperty(keyBuilder.controllerMessage(message.getMsgId()));
+ accessor.removeProperty(keyBuilder.controllerMessage(message.getMsgIdString()));
} else {
- accessor.removeProperty(keyBuilder.message(_manager.getInstanceName(), message.getMsgId()));
+ accessor.removeProperty(keyBuilder.message(_manager.getInstanceName(), message.getMsgIdString()));
}
}
@@ -214,11 +214,11 @@ public class HelixTask implements MessageTask {
if (message.getSrcInstanceType() == InstanceType.PARTICIPANT) {
Builder keyBuilder = accessor.keyBuilder();
- accessor.setProperty(keyBuilder.message(message.getMsgSrc(), replyMessage.getMsgId()),
+ accessor.setProperty(keyBuilder.message(message.getMsgSrc(), replyMessage.getMsgIdString()),
replyMessage);
} else if (message.getSrcInstanceType() == InstanceType.CONTROLLER) {
Builder keyBuilder = accessor.keyBuilder();
- accessor.setProperty(keyBuilder.controllerMessage(replyMessage.getMsgId()), replyMessage);
+ accessor.setProperty(keyBuilder.controllerMessage(replyMessage.getMsgIdString()), replyMessage);
}
_statusUpdateUtil.logInfo(message, HelixTask.class,
"1 msg replied to " + replyMessage.getTgtName(), accessor);
@@ -237,8 +237,8 @@ public class HelixTask implements MessageTask {
long totalDelay = now - msgReadTime;
long executionDelay = now - msgExecutionStartTime;
if (totalDelay > 0 && executionDelay > 0) {
- String fromState = message.getFromState();
- String toState = message.getToState();
+ String fromState = message.getFromStateString();
+ String toState = message.getToStateString();
String transition = fromState + "--" + toState;
StateTransitionContext cxt =
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
index 600a3ab..276c27f 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
@@ -430,14 +430,14 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
continue;
}
- String tgtSessionId = message.getTgtSessionId();
+ String tgtSessionId = message.getTgtSessionIdString();
// sessionId mismatch normally means message comes from expired session, just remove it
if (!sessionId.equals(tgtSessionId) && !tgtSessionId.equals("*")) {
String warningMessage =
"SessionId does NOT match. expected sessionId: " + sessionId
+ ", tgtSessionId in message: " + tgtSessionId + ", messageId: "
- + message.getMsgId();
+ + message.getMsgIdString();
LOG.warn(warningMessage);
accessor.removeProperty(message.getKey(keyBuilder, instanceName));
_statusUpdateUtil.logWarning(message, HelixStateMachineEngine.class, warningMessage,
@@ -452,7 +452,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
// We will read the message again if there is a new message but we
// check for the status and ignore if its already read
if (LOG.isTraceEnabled()) {
- LOG.trace("Message already read. msgId: " + message.getMsgId());
+ LOG.trace("Message already read. msgId: " + message.getMsgIdString());
}
continue;
}
@@ -465,9 +465,9 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
}
handlers.add(createHandler);
} catch (Exception e) {
- LOG.error("Failed to create message handler for " + message.getMsgId(), e);
+ LOG.error("Failed to create message handler for " + message.getMsgIdString(), e);
String error =
- "Failed to create message handler for " + message.getMsgId() + ", exception: " + e;
+ "Failed to create message handler for " + message.getMsgIdString() + ", exception: " + e;
_statusUpdateUtil.logError(message, HelixStateMachineEngine.class, e, error, accessor);
@@ -543,7 +543,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
// the corresponding MessageHandlerFactory is registered
if (handlerFactory == null) {
LOG.warn("Fail to find message handler factory for type: " + msgType + " msgId: "
- + message.getMsgId());
+ + message.getMsgIdString());
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageTimeoutTask.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageTimeoutTask.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageTimeoutTask.java
index e1b4f0f..17f525e 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageTimeoutTask.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageTimeoutTask.java
@@ -40,7 +40,7 @@ public class MessageTimeoutTask extends TimerTask {
Message message = _task.getMessage();
// NotificationContext context = _task.getNotificationContext();
// System.out.println("msg: " + message.getMsgId() + " timeouot.");
- LOG.warn("Message time out, canceling. id:" + message.getMsgId() + " timeout : "
+ LOG.warn("Message time out, canceling. id:" + message.getMsgIdString() + " timeout : "
+ message.getExecutionTimeout());
_task.onTimeout();
_executor.cancelTask(_task);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/ClusterConstraints.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/ClusterConstraints.java b/helix-core/src/main/java/org/apache/helix/model/ClusterConstraints.java
index f69a7ce..3b7557d 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ClusterConstraints.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ClusterConstraints.java
@@ -164,8 +164,8 @@ public class ClusterConstraints extends HelixProperty {
String msgType = msg.getMsgType();
attributes.put(ConstraintAttribute.MESSAGE_TYPE, msgType);
if (MessageType.STATE_TRANSITION.toString().equals(msgType)) {
- if (msg.getFromState() != null && msg.getToState() != null) {
- attributes.put(ConstraintAttribute.TRANSITION, msg.getFromState() + "-" + msg.getToState());
+ if (msg.getFromStateString() != null && msg.getToStateString() != null) {
+ attributes.put(ConstraintAttribute.TRANSITION, msg.getFromStateString() + "-" + msg.getToStateString());
}
if (msg.getResourceName() != null) {
attributes.put(ConstraintAttribute.RESOURCE, msg.getResourceName());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/CurrentState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/CurrentState.java b/helix-core/src/main/java/org/apache/helix/model/CurrentState.java
index 32854ab..dba439b 100644
--- a/helix-core/src/main/java/org/apache/helix/model/CurrentState.java
+++ b/helix-core/src/main/java/org/apache/helix/model/CurrentState.java
@@ -25,6 +25,12 @@ import java.util.TreeMap;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.ResourceId;
+import org.apache.helix.api.SessionId;
+import org.apache.helix.api.State;
+import org.apache.helix.api.StateModelDefId;
import org.apache.log4j.Logger;
/**
@@ -70,10 +76,18 @@ public class CurrentState extends HelixProperty {
}
/**
+ * Get the resource id
+ * @return ResourceId
+ */
+ public ResourceId getResourceId() {
+ return Id.resource(getResourceName());
+ }
+
+ /**
* Get the partitions on this instance and the state that each partition is currently in.
* @return (partition, state) pairs
*/
- public Map<String, String> getPartitionStateMap() {
+ public Map<String, String> getPartitionStateStringMap() {
Map<String, String> map = new HashMap<String, String>();
Map<String, Map<String, String>> mapFields = _record.getMapFields();
for (String partitionName : mapFields.keySet()) {
@@ -86,14 +100,38 @@ public class CurrentState extends HelixProperty {
}
/**
+ * Get the partitions on this instance and the state that each partition is currently in
+ * @return (partition id, state) pairs
+ */
+ public Map<PartitionId, State> getPartitionStateMap() {
+ Map<PartitionId, State> map = new HashMap<PartitionId, State>();
+ for (String partitionName : _record.getMapFields().keySet()) {
+ Map<String, String> stateMap = _record.getMapField(partitionName);
+ if (stateMap != null) {
+ map.put(Id.partition(partitionName),
+ State.from(stateMap.get(CurrentStateProperty.CURRENT_STATE.toString())));
+ }
+ }
+ return map;
+ }
+
+ /**
* Get the session that this current state corresponds to
* @return String session identifier
*/
- public String getSessionId() {
+ public String getSessionIdString() {
return _record.getSimpleField(CurrentStateProperty.SESSION_ID.toString());
}
/**
+ * Get the session that this current state corresponds to
+ * @return session identifier
+ */
+ public SessionId getSessionId() {
+ return Id.session(getSessionIdString());
+ }
+
+ /**
* Set the session that this current state corresponds to
* @param sessionId String session identifier
*/
@@ -116,6 +154,15 @@ public class CurrentState extends HelixProperty {
}
/**
+ * Get the state of a partition on this instance
+ * @param partitionId partition id
+ * @return State
+ */
+ public State getState(PartitionId partitionId) {
+ return State.from(getState(partitionId.stringify()));
+ }
+
+ /**
* Set the state model that the resource follows
* @param stateModelName an identifier of the state model
*/
@@ -132,6 +179,14 @@ public class CurrentState extends HelixProperty {
}
/**
+ * Get the state model that the resource follows
+ * @return an identifier of the state model
+ */
+ public StateModelDefId getStateModelDefId() {
+ return Id.stateModelDef(getStateModelDefRef());
+ }
+
+ /**
* Set the state that a partition is currently in on this instance
* @param partitionName the name of the partition
* @param state the state of the partition
@@ -195,7 +250,7 @@ public class CurrentState extends HelixProperty {
LOG.error("Current state does not contain state model ref. id:" + getResourceName());
return false;
}
- if (getSessionId() == null) {
+ if (getSessionIdString() == null) {
LOG.error("CurrentState does not contain session id, id : " + getResourceName());
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/ExternalView.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/ExternalView.java b/helix-core/src/main/java/org/apache/helix/model/ExternalView.java
index d5f1afc..c7ff8bb 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ExternalView.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ExternalView.java
@@ -25,6 +25,14 @@ import java.util.TreeMap;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ParticipantId;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.ResourceId;
+import org.apache.helix.api.State;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
/**
* External view is an aggregation (across all instances)
@@ -74,11 +82,23 @@ public class ExternalView extends HelixProperty {
* Get all the partitions of the resource
* @return a set of partition names
*/
- public Set<String> getPartitionSet() {
+ public Set<String> getPartitionStringSet() {
return _record.getMapFields().keySet();
}
/**
+ * Get all the partitions of the resource
+ * @return a set of partition ids
+ */
+ public Set<PartitionId> getPartitionSet() {
+ ImmutableSet.Builder<PartitionId> builder = new ImmutableSet.Builder<PartitionId>();
+ for (String partitionName : getPartitionStringSet()) {
+ builder.add(Id.partition(partitionName));
+ }
+ return builder.build();
+ }
+
+ /**
* Get the instance and the state for each partition replica
* @param partitionName the partition to look up
* @return (instance, state) pairs
@@ -88,6 +108,21 @@ public class ExternalView extends HelixProperty {
}
/**
+ * Get the participant and the state for each partition replica
+ * @param partitionId the partition to look up
+ * @return (participant, state) pairs
+ */
+ public Map<ParticipantId, State> getStateMap(PartitionId partitionId) {
+ Map<String, String> rawStateMap = getStateMap(partitionId.stringify());
+ ImmutableMap.Builder<ParticipantId, State> builder =
+ new ImmutableMap.Builder<ParticipantId, State>();
+ for (String participantName : rawStateMap.keySet()) {
+ builder.put(Id.participant(participantName), State.from(rawStateMap.get(participantName)));
+ }
+ return builder.build();
+ }
+
+ /**
* Get the resource represented by this view
* @return the name of the resource
*/
@@ -95,6 +130,14 @@ public class ExternalView extends HelixProperty {
return _record.getId();
}
+ /**
+ * Get the resource represented by this view
+ * @return resource id
+ */
+ public ResourceId getResourceId() {
+ return Id.resource(getResourceName());
+ }
+
@Override
public boolean isValid() {
return true;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/IdealState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/IdealState.java b/helix-core/src/main/java/org/apache/helix/model/IdealState.java
index e14940a..cbf7a64 100644
--- a/helix-core/src/main/java/org/apache/helix/model/IdealState.java
+++ b/helix-core/src/main/java/org/apache/helix/model/IdealState.java
@@ -30,9 +30,20 @@ import java.util.TreeSet;
import org.apache.helix.HelixConstants;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ParticipantId;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.RebalancerRef;
+import org.apache.helix.api.ResourceId;
+import org.apache.helix.api.State;
+import org.apache.helix.api.StateModelDefId;
import org.apache.helix.controller.rebalancer.Rebalancer;
import org.apache.log4j.Logger;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
/**
* The ideal states of all partitions in a resource
*/
@@ -107,6 +118,14 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get the associated resource
+ * @return the id of the resource
+ */
+ public ResourceId getResourceId() {
+ return Id.resource(getResourceName());
+ }
+
+ /**
* Get the rebalance mode of the ideal state
* @param mode {@link IdealStateModeProperty}
*/
@@ -154,6 +173,18 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get a reference to the user-defined rebalancer associated with this resource(if any)
+ * @return
+ */
+ public RebalancerRef getRebalancerRef() {
+ String className = getRebalancerClassName();
+ if (className != null) {
+ return RebalancerRef.from(getRebalancerClassName());
+ }
+ return null;
+ }
+
+ /**
* Set the maximum number of partitions of this resource that an instance can serve
* @param max the maximum number of partitions supported
*/
@@ -204,7 +235,7 @@ public class IdealState extends HelixProperty {
* Get all of the partitions
* @return a set of partition names
*/
- public Set<String> getPartitionSet() {
+ public Set<String> getPartitionStringSet() {
if (getRebalanceMode() == RebalanceMode.SEMI_AUTO
|| getRebalanceMode() == RebalanceMode.FULL_AUTO) {
return _record.getListFields().keySet();
@@ -218,6 +249,18 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get all of the partitions
+ * @return an immutable set of partitions
+ */
+ public Set<PartitionId> getPartitionSet() {
+ ImmutableSet.Builder<PartitionId> partitionSetBuilder = new ImmutableSet.Builder<PartitionId>();
+ for (String partitionName : getPartitionStringSet()) {
+ partitionSetBuilder.add(Id.partition(partitionName));
+ }
+ return partitionSetBuilder.build();
+ }
+
+ /**
* Set the current mapping of a partition
* @param partition the partition to set
* @param instanceStateMap (instance name, state) pairs
@@ -236,6 +279,21 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get the current mapping of a partition
+ * @param partitionId the name of the partition
+ * @return the instances where the replicas live and the state of each (immutable)
+ */
+ public Map<ParticipantId, State> getParticipantStateMap(PartitionId partitionId) {
+ Map<String, String> instanceStateMap = getInstanceStateMap(partitionId.stringify());
+ ImmutableMap.Builder<ParticipantId, State> builder =
+ new ImmutableMap.Builder<ParticipantId, State>();
+ for (String participantId : instanceStateMap.keySet()) {
+ builder.put(Id.participant(participantId), State.from(instanceStateMap.get(participantId)));
+ }
+ return builder.build();
+ }
+
+ /**
* Get the instances who host replicas of a partition
* @param partitionName the partition to look up
* @return set of instance names
@@ -263,7 +321,19 @@ public class IdealState extends HelixProperty {
logger.error("Invalid ideal state mode: " + getResourceName());
return Collections.emptySet();
}
+ }
+ /**
+ * Get the participants who host replicas of a partition
+ * @param partitionId the partition to look up
+ * @return immutable set of participant ids
+ */
+ public Set<ParticipantId> getParticipantSet(PartitionId partitionId) {
+ ImmutableSet.Builder<ParticipantId> builder = new ImmutableSet.Builder<ParticipantId>();
+ for (String participantName : getInstanceSet(partitionId.stringify())) {
+ builder.add(Id.participant(participantName));
+ }
+ return builder.build();
}
/**
@@ -291,6 +361,20 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get the preference list of a partition
+ * @param partitionId the partition id
+ * @return an ordered list of participants that can serve replicas of the partition
+ */
+ public List<ParticipantId> getPreferenceList(PartitionId partitionId) {
+ ImmutableList.Builder<ParticipantId> builder = new ImmutableList.Builder<ParticipantId>();
+ List<String> preferenceStringList = getPreferenceList(partitionId.stringify());
+ for (String participantName : preferenceStringList) {
+ builder.add(Id.participant(participantName));
+ }
+ return builder.build();
+ }
+
+ /**
* Get the state model associated with this resource
* @return an identifier of the state model
*/
@@ -299,6 +383,14 @@ public class IdealState extends HelixProperty {
}
/**
+ * Get the state model associated with this resource
+ * @return an identifier of the state model
+ */
+ public StateModelDefId getStateModelDefId() {
+ return Id.stateModelDef(getStateModelDefRef());
+ }
+
+ /**
* Set the state model associated with this resource
* @param stateModel state model identifier
*/
@@ -426,7 +518,7 @@ public class IdealState extends HelixProperty {
if (!replicaStr.equals(HelixConstants.StateModelToken.ANY_LIVEINSTANCE.toString())) {
int replica = Integer.parseInt(replicaStr);
- Set<String> partitionSet = getPartitionSet();
+ Set<String> partitionSet = getPartitionStringSet();
for (String partition : partitionSet) {
List<String> preferenceList = getPreferenceList(partition);
if (preferenceList == null || preferenceList.size() != replica) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
index eb1c652..2905965 100644
--- a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
@@ -27,6 +27,8 @@ import java.util.Set;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ParticipantId;
import org.apache.log4j.Logger;
/**
@@ -238,6 +240,14 @@ public class InstanceConfig extends HelixProperty {
return _record.getId();
}
+ /**
+ * Get the identifier of this participant
+ * @return the participant id
+ */
+ public ParticipantId getParticipantId() {
+ return Id.participant(getInstanceName());
+ }
+
@Override
public boolean isValid() {
// HELIX-65: remove check for hostname/port existence
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/433b0011/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
index 75e0cf3..86eb2d8 100644
--- a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
+++ b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
@@ -21,6 +21,11 @@ package org.apache.helix.model;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.HelixVersion;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ParticipantId;
+import org.apache.helix.api.ProcId;
+import org.apache.helix.api.SessionId;
import org.apache.log4j.Logger;
/**
@@ -67,11 +72,19 @@ public class LiveInstance extends HelixProperty {
* Get the session that this instance corresponds to
* @return session identifier
*/
- public String getSessionId() {
+ public String getSessionIdString() {
return _record.getSimpleField(LiveInstanceProperty.SESSION_ID.toString());
}
/**
+ * Get the session that this participant corresponds to
+ * @return session identifier
+ */
+ public SessionId getSessionId() {
+ return Id.session(getSessionIdString());
+ }
+
+ /**
* Get the name of this instance
* @return the instance name
*/
@@ -80,14 +93,30 @@ public class LiveInstance extends HelixProperty {
}
/**
+ * Get the id of this participant
+ * @return participant id
+ */
+ public ParticipantId getParticipantId() {
+ return Id.participant(getInstanceName());
+ }
+
+ /**
* Get the version of Helix that this instance is running
* @return the version
*/
- public String getHelixVersion() {
+ public String getHelixVersionString() {
return _record.getSimpleField(LiveInstanceProperty.HELIX_VERSION.toString());
}
/**
+ * Get the version of Helix that this participant is running
+ * @return the version
+ */
+ public HelixVersion getHelixVersion() {
+ return HelixVersion.from(getHelixVersionString());
+ }
+
+ /**
* Set the version of Helix that this instance is running
* @param helixVersion the version
*/
@@ -104,6 +133,14 @@ public class LiveInstance extends HelixProperty {
}
/**
+ * Get an identifier that represents the instance and where it is located
+ * @return process identifier
+ */
+ public ProcId getProcessId() {
+ return Id.process(getLiveInstance());
+ }
+
+ /**
* Set an identifier that represents the process
* @param liveInstance process identifier, e.g. process_id@host
*/
@@ -137,11 +174,11 @@ public class LiveInstance extends HelixProperty {
@Override
public boolean isValid() {
- if (getSessionId() == null) {
+ if (getSessionIdString() == null) {
_logger.error("liveInstance does not have session id. id:" + _record.getId());
return false;
}
- if (getHelixVersion() == null) {
+ if (getHelixVersionString() == null) {
_logger.error("liveInstance does not have CLM verion. id:" + _record.getId());
return false;
}