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/30 19:16:29 UTC
[3/5] [HELIX-109] Review Helix model package, rb=13878
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/model/Message.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/Message.java b/helix-core/src/main/java/org/apache/helix/model/Message.java
index 5bc4a46..b2b538d 100644
--- a/helix-core/src/main/java/org/apache/helix/model/Message.java
+++ b/helix-core/src/main/java/org/apache/helix/model/Message.java
@@ -120,7 +120,7 @@ public class Message extends HelixProperty {
* @param type the message category
* @param msgId unique message identifier
*/
- public Message(MessageType type, String msgId) {
+ public Message(MessageType type, MessageId msgId) {
this(type.toString(), msgId);
}
@@ -129,8 +129,8 @@ public class Message extends HelixProperty {
* @param type {@link MessageType} as a string or a custom message type
* @param msgId unique message identifier
*/
- public Message(String type, String msgId) {
- super(new ZNRecord(msgId));
+ public Message(String type, MessageId msgId) {
+ super(new ZNRecord(msgId.stringify()));
_record.setSimpleField(Attributes.MSG_TYPE.toString(), type);
setMsgId(msgId);
setMsgState(MessageState.NEW);
@@ -138,6 +138,16 @@ public class Message extends HelixProperty {
}
/**
+ * Instantiate a message with a new id
+ * @param record a ZNRecord corresponding to a message
+ * @param id unique message identifier
+ */
+ public Message(ZNRecord record, MessageId id) {
+ super(new ZNRecord(record, id.stringify()));
+ setMsgId(id);
+ }
+
+ /**
* Instantiate a message
* @param record a ZNRecord corresponding to a message
*/
@@ -160,16 +170,6 @@ public class Message extends HelixProperty {
}
/**
- * Instantiate a message with a new id
- * @param record a ZNRecord corresponding to a message
- * @param id unique message identifier
- */
- public Message(ZNRecord record, String id) {
- super(new ZNRecord(record, id));
- setMsgId(id);
- }
-
- /**
* Set a subtype of the message
* @param subType name of the subtype
*/
@@ -205,32 +205,16 @@ public class Message extends HelixProperty {
* Get the session identifier of the destination node
* @return session identifier
*/
- public String getTgtSessionIdString() {
- return _record.getSimpleField(Attributes.TGT_SESSION_ID.toString());
- }
-
- /**
- * Get the session identifier of the destination node
- * @return session identifier
- */
public SessionId getTgtSessionId() {
- return Id.session(getTgtSessionIdString());
+ return Id.session(_record.getSimpleField(Attributes.TGT_SESSION_ID.toString()));
}
/**
* Set the session identifier of the destination node
* @param tgtSessionId session identifier
*/
- public void setTgtSessionId(String tgtSessionId) {
- _record.setSimpleField(Attributes.TGT_SESSION_ID.toString(), tgtSessionId);
- }
-
- /**
- * Get the session identifier of the source node
- * @return session identifier
- */
- public String getSrcSessionIdString() {
- return _record.getSimpleField(Attributes.SRC_SESSION_ID.toString());
+ public void setTgtSessionId(SessionId tgtSessionId) {
+ _record.setSimpleField(Attributes.TGT_SESSION_ID.toString(), tgtSessionId.stringify());
}
/**
@@ -238,23 +222,15 @@ public class Message extends HelixProperty {
* @return session identifier
*/
public SessionId getSrcSessionId() {
- return Id.session(getSrcSessionIdString());
+ return Id.session(_record.getSimpleField(Attributes.SRC_SESSION_ID.toString()));
}
/**
* Set the session identifier of the source node
* @param srcSessionId session identifier
*/
- public void setSrcSessionId(String srcSessionId) {
- _record.setSimpleField(Attributes.SRC_SESSION_ID.toString(), srcSessionId);
- }
-
- /**
- * Get the session identifier of the node that executes the message
- * @return session identifier
- */
- public String getExecutionSessionIdString() {
- return _record.getSimpleField(Attributes.EXE_SESSION_ID.toString());
+ public void setSrcSessionId(SessionId srcSessionId) {
+ _record.setSimpleField(Attributes.SRC_SESSION_ID.toString(), srcSessionId.stringify());
}
/**
@@ -262,19 +238,15 @@ public class Message extends HelixProperty {
* @return session identifier
*/
public SessionId getExecutionSessionId() {
- String sessionId = getExecutionSessionIdString();
- if (sessionId != null) {
- return Id.session(sessionId);
- }
- return null;
+ return Id.session(_record.getSimpleField(Attributes.EXE_SESSION_ID.toString()));
}
/**
* Set the session identifier of the node that executes the message
* @param exeSessionId session identifier
*/
- public void setExecuteSessionId(String exeSessionId) {
- _record.setSimpleField(Attributes.EXE_SESSION_ID.toString(), exeSessionId);
+ public void setExecuteSessionId(SessionId exeSessionId) {
+ _record.setSimpleField(Attributes.EXE_SESSION_ID.toString(), exeSessionId.stringify());
}
/**
@@ -338,19 +310,11 @@ public class Message extends HelixProperty {
}
/**
- * Set the name of the partition this message concerns
- * @param partitionName
+ * Set the id of the partition this message concerns
+ * @param partitionId
*/
- public void setPartitionName(String partitionName) {
- _record.setSimpleField(Attributes.PARTITION_NAME.toString(), partitionName);
- }
-
- /**
- * Get the unique identifier of this message
- * @return message identifier
- */
- public String getMsgIdString() {
- return _record.getSimpleField(Attributes.MSG_ID.toString());
+ public void setPartitionId(PartitionId partitionId) {
+ _record.setSimpleField(Attributes.PARTITION_NAME.toString(), partitionId.stringify());
}
/**
@@ -358,31 +322,23 @@ public class Message extends HelixProperty {
* @return message identifier
*/
public MessageId getMsgId() {
- return Id.message(getMsgIdString());
+ return Id.message(_record.getSimpleField(Attributes.MSG_ID.toString()));
}
/**
* Set the unique identifier of this message
* @param msgId message identifier
*/
- public void setMsgId(String msgId) {
- _record.setSimpleField(Attributes.MSG_ID.toString(), msgId);
+ public void setMsgId(MessageId msgId) {
+ _record.setSimpleField(Attributes.MSG_ID.toString(), msgId.toString());
}
/**
* Set the "from state" for transition-related messages
- * @param state the state name
- */
- public void setFromState(String state) {
- _record.setSimpleField(Attributes.FROM_STATE.toString(), state);
- }
-
- /**
- * Get the "from-state" for transition-related messages
- * @return state name, or null for other message types
+ * @param state the state
*/
- public String getFromStateString() {
- return _record.getSimpleField(Attributes.FROM_STATE.toString());
+ public void setFromState(State state) {
+ _record.setSimpleField(Attributes.FROM_STATE.toString(), state.toString());
}
/**
@@ -390,23 +346,15 @@ public class Message extends HelixProperty {
* @return state, or null for other message types
*/
public State getFromState() {
- return State.from(getFromStateString());
+ return State.from(_record.getSimpleField(Attributes.FROM_STATE.toString()));
}
/**
* Set the "to state" for transition-related messages
- * @param state the state name
+ * @param state the state
*/
- public void setToState(String state) {
- _record.setSimpleField(Attributes.TO_STATE.toString(), state);
- }
-
- /**
- * Get the "to state" for transition-related messages
- * @return state name, or null for other message types
- */
- public String getToStateString() {
- return _record.getSimpleField(Attributes.TO_STATE.toString());
+ public void setToState(State state) {
+ _record.setSimpleField(Attributes.TO_STATE.toString(), state.toString());
}
/**
@@ -414,7 +362,7 @@ public class Message extends HelixProperty {
* @return state, or null for other message types
*/
public State getToState() {
- return State.from(getToStateString());
+ return State.from(_record.getSimpleField(Attributes.TO_STATE.toString()));
}
/**
@@ -443,34 +391,18 @@ public class Message extends HelixProperty {
/**
* Set the resource associated with this message
- * @param resourceName resource name to set
+ * @param resourceId resource name to set
*/
- public void setResourceName(String resourceName) {
- _record.setSimpleField(Attributes.RESOURCE_NAME.toString(), resourceName);
+ public void setResourceId(ResourceId resourceId) {
+ _record.setSimpleField(Attributes.RESOURCE_NAME.toString(), resourceId.stringify());
}
/**
* Get the resource associated with this message
* @return resource name
*/
- public String getResourceName() {
- return _record.getSimpleField(Attributes.RESOURCE_NAME.toString());
- }
-
- /**
- * Get the resource associated with this message
- * @return resource id
- */
public ResourceId getResourceId() {
- return Id.resource(getResourceName());
- }
-
- /**
- * Get the resource partition associated with this message
- * @return partition name
- */
- public String getPartitionName() {
- return _record.getSimpleField(Attributes.PARTITION_NAME.toString());
+ return Id.resource(_record.getSimpleField(Attributes.RESOURCE_NAME.toString()));
}
/**
@@ -478,7 +410,7 @@ public class Message extends HelixProperty {
* @return partition id
*/
public PartitionId getPartitionId() {
- return Id.partition(getPartitionName());
+ return Id.partition(_record.getSimpleField(Attributes.PARTITION_NAME.toString()));
}
/**
@@ -498,11 +430,11 @@ public class Message extends HelixProperty {
}
/**
- * Set the state model definition name
+ * Set the state model definition
* @param stateModelDefName a reference to the state model definition, e.g. "MasterSlave"
*/
- public void setStateModelDef(String stateModelDefName) {
- _record.setSimpleField(Attributes.STATE_MODEL_DEF.toString(), stateModelDefName);
+ public void setStateModelDef(StateModelDefId stateModelDefId) {
+ _record.setSimpleField(Attributes.STATE_MODEL_DEF.toString(), stateModelDefId.stringify());
}
/**
@@ -666,13 +598,14 @@ public class Message extends HelixProperty {
public static Message createReplyMessage(Message srcMessage, String instanceName,
Map<String, String> taskResultMap) {
if (srcMessage.getCorrelationId() == null) {
- throw new HelixException("Message " + srcMessage.getMsgIdString()
+ throw new HelixException("Message " + srcMessage.getMsgId()
+ " does not contain correlation id");
}
- Message replyMessage = new Message(MessageType.TASK_REPLY, UUID.randomUUID().toString());
+ Message replyMessage =
+ new Message(MessageType.TASK_REPLY, Id.message(UUID.randomUUID().toString()));
replyMessage.setCorrelationId(srcMessage.getCorrelationId());
replyMessage.setResultMap(taskResultMap);
- replyMessage.setTgtSessionId("*");
+ replyMessage.setTgtSessionId(Id.session("*"));
replyMessage.setMsgState(MessageState.NEW);
replyMessage.setSrcName(instanceName);
if (srcMessage.getSrcInstanceType() == InstanceType.CONTROLLER) {
@@ -700,24 +633,15 @@ public class Message extends HelixProperty {
/**
* Get a list of partitions associated with this message
- * @return list of partition names
+ * @return list of partition ids
*/
- public List<String> getPartitionNames() {
+ public List<PartitionId> getPartitionIds() {
List<String> partitionNames = _record.getListField(Attributes.PARTITION_NAME.toString());
if (partitionNames == null) {
return Collections.emptyList();
}
-
- return partitionNames;
- }
-
- /**
- * Get a list of partitions associated with this message
- * @return list of partition ids
- */
- public List<PartitionId> getPartitionIds() {
ImmutableList.Builder<PartitionId> builder = new ImmutableList.Builder<PartitionId>();
- for (String partitionName : getPartitionNames()) {
+ for (String partitionName : partitionNames) {
builder.add(Id.partition(partitionName));
}
return builder.build();
@@ -766,10 +690,11 @@ public class Message extends HelixProperty {
if (getMsgType().equals(MessageType.STATE_TRANSITION.toString())) {
boolean isNotValid =
- isNullOrEmpty(getTgtName()) || isNullOrEmpty(getPartitionName())
- || isNullOrEmpty(getResourceName()) || isNullOrEmpty(getStateModelDef())
- || isNullOrEmpty(getToStateString()) || isNullOrEmpty(getStateModelFactoryName())
- || isNullOrEmpty(getFromStateString());
+ isNullOrEmpty(getTgtName()) || isNullOrEmpty(getPartitionId().stringify())
+ || isNullOrEmpty(getResourceId().stringify()) || isNullOrEmpty(getStateModelDef())
+ || isNullOrEmpty(getToState().toString())
+ || isNullOrEmpty(getStateModelFactoryName())
+ || isNullOrEmpty(getFromState().toString());
return !isNotValid;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/model/ResourceAssignment.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/ResourceAssignment.java b/helix-core/src/main/java/org/apache/helix/model/ResourceAssignment.java
index 7013829..bbc2baa 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ResourceAssignment.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ResourceAssignment.java
@@ -19,12 +19,20 @@ package org.apache.helix.model;
* under the License.
*/
-import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.helix.HelixProperty;
+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.ImmutableList;
+import com.google.common.collect.ImmutableMap;
/**
* Represents the assignments of replicas for an entire resource, keyed on partitions of the
@@ -43,8 +51,8 @@ public class ResourceAssignment extends HelixProperty {
* Initialize an empty mapping
* @param resourceName the resource being mapped
*/
- public ResourceAssignment(String resourceName) {
- super(resourceName);
+ public ResourceAssignment(ResourceId resourceId) {
+ super(resourceId.stringify());
}
/**
@@ -57,42 +65,79 @@ public class ResourceAssignment extends HelixProperty {
/**
* Get the resource for which this assignment was created
- * @return resource name
+ * @return resource id
*/
- public String getResourceName() {
- return getId();
+ public ResourceId getResourceId() {
+ return Id.resource(getId());
}
/**
* Get the currently mapped partitions
* @return list of Partition objects
*/
- public List<Partition> getMappedPartitions() {
- List<Partition> partitions = new ArrayList<Partition>();
+ public List<PartitionId> getMappedPartitions() {
+ ImmutableList.Builder<PartitionId> builder = new ImmutableList.Builder<PartitionId>();
for (String partitionName : _record.getMapFields().keySet()) {
- partitions.add(new Partition(partitionName));
+ builder.add(Id.partition(partitionName));
}
- return partitions;
+ return builder.build();
}
/**
- * Get the instance, state pairs for a partition
+ * Get the participant, state pairs for a partition
* @param partition the Partition to look up
- * @return map of (instance name, state)
+ * @return map of (participant id, state)
+ */
+ public Map<ParticipantId, State> getReplicaMap(PartitionId partitionId) {
+ Map<String, String> rawReplicaMap = _record.getMapField(partitionId.stringify());
+ if (rawReplicaMap == null) {
+ return Collections.emptyMap();
+ }
+ ImmutableMap.Builder<ParticipantId, State> builder =
+ new ImmutableMap.Builder<ParticipantId, State>();
+ for (String participantName : rawReplicaMap.keySet()) {
+ builder.put(Id.participant(participantName), State.from(rawReplicaMap.get(participantName)));
+ }
+ return builder.build();
+ }
+
+ /**
+ * Add participant, state pairs for a partition
+ * TODO: should be package-private, but builder can't see it
+ * @param partitionId the partition to set
+ * @param replicaMap map of (participant name, state)
*/
- public Map<String, String> getReplicaMap(Partition partition) {
- if (_record.getMapFields().containsKey(partition.getPartitionName())) {
- return _record.getMapField(partition.getPartitionName());
+ public void addReplicaMap(PartitionId partitionId, Map<ParticipantId, State> replicaMap) {
+ Map<String, String> convertedMap = new HashMap<String, String>();
+ for (ParticipantId participantId : replicaMap.keySet()) {
+ convertedMap.put(participantId.stringify(), replicaMap.get(participantId).toString());
}
- return Collections.emptyMap();
+ _record.setMapField(partitionId.stringify(), convertedMap);
}
/**
- * Add instance, state pairs for a partition
- * @param partition the partition to set
- * @param replicaMap map of (instance name, state)
+ * Helper for converting a map of strings to a concrete replica map
+ * @param rawMap map of participant name to state name
+ * @return map of participant id to state
*/
- public void addReplicaMap(Partition partition, Map<String, String> replicaMap) {
- _record.setMapField(partition.getPartitionName(), replicaMap);
+ public static Map<ParticipantId, State> replicaMapFromStringMap(Map<String, String> rawMap) {
+ Map<ParticipantId, State> replicaMap = new HashMap<ParticipantId, State>();
+ for (String participantName : rawMap.keySet()) {
+ replicaMap.put(Id.participant(participantName), State.from(rawMap.get(participantName)));
+ }
+ return replicaMap;
+ }
+
+ /**
+ * Helper for converting a replica map to a map of strings
+ * @param replicaMap map of participant id to state
+ * @return map of participant name to state name
+ */
+ public static Map<String, String> stringMapFromReplicaMap(Map<ParticipantId, State> replicaMap) {
+ Map<String, String> rawMap = new HashMap<String, String>();
+ for (ParticipantId participantId : replicaMap.keySet()) {
+ rawMap.put(participantId.stringify(), replicaMap.get(participantId).toString());
+ }
+ return rawMap;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/model/StateModelDefinition.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/StateModelDefinition.java b/helix-core/src/main/java/org/apache/helix/model/StateModelDefinition.java
index caaf81e..3ca9776 100644
--- a/helix-core/src/main/java/org/apache/helix/model/StateModelDefinition.java
+++ b/helix-core/src/main/java/org/apache/helix/model/StateModelDefinition.java
@@ -318,7 +318,7 @@ public class StateModelDefinition extends HelixProperty {
* @return Builder
*/
public Builder addTransition(String fromState, String toState, int priority) {
- transitionMap.put(new Transition(fromState, toState), priority);
+ transitionMap.put(new Transition(State.from(fromState), State.from(toState)), priority);
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/model/Transition.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/Transition.java b/helix-core/src/main/java/org/apache/helix/model/Transition.java
index 0e7f32f..4d4bac9 100644
--- a/helix-core/src/main/java/org/apache/helix/model/Transition.java
+++ b/helix-core/src/main/java/org/apache/helix/model/Transition.java
@@ -1,5 +1,7 @@
package org.apache.helix.model;
+import org.apache.helix.api.State;
+
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -23,15 +25,15 @@ package org.apache.helix.model;
* Defines a transition from one state to another
*/
public class Transition {
- final private String _fromState;
- final private String _toState;
+ final private State _fromState;
+ final private State _toState;
/**
* Instantiate with a source and destination state
* @param fromState source name
* @param toState destination name
*/
- public Transition(String fromState, String toState) {
+ public Transition(State fromState, State toState) {
_fromState = fromState;
_toState = toState;
}
@@ -58,7 +60,7 @@ public class Transition {
* Get the source state
* @return source state name
*/
- public String getFromState() {
+ public State getFromState() {
return _fromState;
}
@@ -66,7 +68,7 @@ public class Transition {
* Get the destination state
* @return destination state name
*/
- public String getToState() {
+ public State getToState() {
return _toState;
}
@@ -78,6 +80,16 @@ public class Transition {
public static Transition from(String transition) {
String fromState = transition.substring(0, transition.indexOf('-'));
String toState = transition.substring(transition.indexOf('-') + 1);
+ return new Transition(State.from(fromState), State.from(toState));
+ }
+
+ /**
+ * Create a new transition
+ * @param fromState string source state
+ * @param toState string destination state
+ * @return Transition
+ */
+ public static Transition from(State fromState, State toState) {
return new Transition(fromState, toState);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/model/builder/StateTransitionTableBuilder.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/builder/StateTransitionTableBuilder.java b/helix-core/src/main/java/org/apache/helix/model/builder/StateTransitionTableBuilder.java
index bbd7b33..e2a3f75 100644
--- a/helix-core/src/main/java/org/apache/helix/model/builder/StateTransitionTableBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/model/builder/StateTransitionTableBuilder.java
@@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.helix.api.State;
import org.apache.helix.model.Transition;
public class StateTransitionTableBuilder {
@@ -125,10 +126,10 @@ public class StateTransitionTableBuilder {
}
for (Transition transition : transitions) {
- String fromState = transition.getFromState();
- String toState = transition.getToState();
- setPathVal(path, fromState, toState, 1);
- setNext(next, fromState, toState, toState);
+ State fromState = transition.getFromState();
+ State toState = transition.getToState();
+ setPathVal(path, fromState.toString(), toState.toString(), 1);
+ setNext(next, fromState.toString(), toState.toString(), toState.toString());
}
// iterate
@@ -162,11 +163,11 @@ public class StateTransitionTableBuilder {
states.add("OFFLINE");
List<Transition> transitions = new ArrayList<Transition>();
- transitions.add(new Transition("SLAVE", "OFFLINE"));
- transitions.add(new Transition("OFFLINE", "SLAVE"));
- transitions.add(new Transition("SLAVE", "MASTER"));
- transitions.add(new Transition("OFFLINE", "DROPPED"));
- transitions.add(new Transition("MASTER", "SLAVE"));
+ transitions.add(Transition.from("SLAVE-OFFLINE"));
+ transitions.add(Transition.from("OFFLINE-SLAVE"));
+ transitions.add(Transition.from("SLAVE-MASTER"));
+ transitions.add(Transition.from("OFFLINE-DROPPED"));
+ transitions.add(Transition.from("MASTER-SLAVE"));
StateTransitionTableBuilder builder = new StateTransitionTableBuilder();
Map<String, Map<String, String>> next = builder.buildTransitionTable(states, transitions);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerElection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerElection.java b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerElection.java
index 3551043..6ae8a1b 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerElection.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerElection.java
@@ -142,7 +142,7 @@ public class DistClusterControllerElection implements ControllerChangeListener {
leader = accessor.getProperty(keyBuilder.controllerLeader());
if (leader != null) {
- String leaderSessionId = leader.getSessionIdString();
+ String leaderSessionId = leader.getSessionId().stringify();
LOG.info("Leader exists for cluster: " + manager.getClusterName() + ", currentLeader: "
+ leader.getInstanceName() + ", leaderSessionId: " + leaderSessionId);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java
index 8c88b7c..0c2eb7c 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/DistClusterControllerStateModel.java
@@ -53,7 +53,7 @@ public class DistClusterControllerStateModel extends StateModel {
@Transition(to = "LEADER", from = "STANDBY")
public void onBecomeLeaderFromStandby(Message message, NotificationContext context)
throws Exception {
- String clusterName = message.getPartitionName();
+ String clusterName = message.getPartitionId().stringify();
String controllerName = message.getTgtName();
logger.info(controllerName + " becomes leader from standby for " + clusterName);
@@ -74,7 +74,7 @@ public class DistClusterControllerStateModel extends StateModel {
@Transition(to = "STANDBY", from = "LEADER")
public void onBecomeStandbyFromLeader(Message message, NotificationContext context) {
- String clusterName = message.getPartitionName();
+ String clusterName = message.getPartitionId().stringify();
String controllerName = message.getTgtName();
logger.info(controllerName + " becoming standby from leader for " + clusterName);
@@ -89,7 +89,7 @@ public class DistClusterControllerStateModel extends StateModel {
@Transition(to = "OFFLINE", from = "STANDBY")
public void onBecomeOfflineFromStandby(Message message, NotificationContext context) {
- String clusterName = message.getPartitionName();
+ String clusterName = message.getPartitionId().stringify();
String controllerName = message.getTgtName();
logger.info(controllerName + " becoming offline from standby for cluster:" + clusterName);
@@ -109,7 +109,7 @@ public class DistClusterControllerStateModel extends StateModel {
@Override
public void rollbackOnError(Message message, NotificationContext context,
StateTransitionError error) {
- String clusterName = message.getPartitionName();
+ String clusterName = message.getPartitionId().stringify();
String controllerName = message.getTgtName();
logger.error(controllerName + " rollbacks on error for " + clusterName);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java b/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
index 5fd276f..31ee954 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
@@ -31,16 +31,21 @@ import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.MapKey;
import org.apache.helix.PropertyKey.Builder;
+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.helix.messaging.handling.BatchMessageHandler;
import org.apache.helix.messaging.handling.BatchMessageWrapper;
import org.apache.helix.messaging.handling.HelixStateTransitionHandler;
-import org.apache.helix.messaging.handling.HelixTaskExecutor;
import org.apache.helix.messaging.handling.MessageHandler;
import org.apache.helix.messaging.handling.TaskExecutor;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
-import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.participant.statemachine.StateModel;
import org.apache.helix.participant.statemachine.StateModelFactory;
import org.apache.helix.participant.statemachine.StateModelParser;
@@ -113,7 +118,7 @@ public class HelixStateMachineEngine implements StateMachineEngine {
private void sendNopMessage() {
if (_manager.isConnected()) {
try {
- Message nopMsg = new Message(MessageType.NO_OP, UUID.randomUUID().toString());
+ Message nopMsg = new Message(MessageType.NO_OP, Id.message(UUID.randomUUID().toString()));
nopMsg.setSrcName(_manager.getInstanceName());
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
@@ -160,16 +165,16 @@ public class HelixStateMachineEngine implements StateMachineEngine {
if (!type.equals(MessageType.STATE_TRANSITION.toString())) {
throw new HelixException("Expect state-transition message type, but was "
- + message.getMsgType() + ", msgId: " + message.getMsgIdString());
+ + message.getMsgType() + ", msgId: " + message.getMsgId());
}
- String partitionKey = message.getPartitionName();
- String stateModelName = message.getStateModelDef();
- String resourceName = message.getResourceName();
- String sessionId = message.getTgtSessionIdString();
+ PartitionId partitionKey = message.getPartitionId();
+ StateModelDefId stateModelId = message.getStateModelDefId();
+ ResourceId resourceId = message.getResourceId();
+ SessionId sessionId = message.getTgtSessionId();
int bucketSize = message.getBucketSize();
- if (stateModelName == null) {
+ if (stateModelId == null) {
logger
.error("Fail to create msg-handler because message does not contain stateModelDef. msgId: "
+ message.getId());
@@ -182,59 +187,61 @@ public class HelixStateMachineEngine implements StateMachineEngine {
}
StateModelFactory<? extends StateModel> stateModelFactory =
- getStateModelFactory(stateModelName, factoryName);
+ getStateModelFactory(stateModelId.stringify(), factoryName);
if (stateModelFactory == null) {
logger.warn("Fail to create msg-handler because cannot find stateModelFactory for model: "
- + stateModelName + " using factoryName: " + factoryName + " for resource: "
- + resourceName);
+ + stateModelId + " using factoryName: " + factoryName + " for resource: " + resourceId);
return null;
}
// check if the state model definition exists and cache it
- if (!_stateModelDefs.containsKey(stateModelName)) {
+ if (!_stateModelDefs.containsKey(stateModelId.stringify())) {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
StateModelDefinition stateModelDef =
- accessor.getProperty(keyBuilder.stateModelDef(stateModelName));
+ accessor.getProperty(keyBuilder.stateModelDef(stateModelId.stringify()));
if (stateModelDef == null) {
throw new HelixException("fail to create msg-handler because stateModelDef for "
- + stateModelName + " does NOT exist");
+ + stateModelId + " does NOT exist");
}
- _stateModelDefs.put(stateModelName, stateModelDef);
+ _stateModelDefs.put(stateModelId.stringify(), stateModelDef);
}
if (message.getBatchMessageMode() == false) {
// create currentStateDelta for this partition
String initState = _stateModelDefs.get(message.getStateModelDef()).getInitialStateString();
- StateModel stateModel = stateModelFactory.getStateModel(partitionKey);
+ StateModel stateModel = stateModelFactory.getStateModel(partitionKey.stringify());
if (stateModel == null) {
- stateModel = stateModelFactory.createAndAddStateModel(partitionKey);
+ stateModel = stateModelFactory.createAndAddStateModel(partitionKey.stringify());
stateModel.updateState(initState);
}
// TODO: move currentStateDelta to StateTransitionMsgHandler
- CurrentState currentStateDelta = new CurrentState(resourceName);
+ CurrentState currentStateDelta = new CurrentState(resourceId.stringify());
currentStateDelta.setSessionId(sessionId);
- currentStateDelta.setStateModelDefRef(stateModelName);
+ currentStateDelta.setStateModelDefRef(stateModelId.stringify());
currentStateDelta.setStateModelFactoryName(factoryName);
currentStateDelta.setBucketSize(bucketSize);
- currentStateDelta.setState(partitionKey, (stateModel.getCurrentState() == null) ? initState
- : stateModel.getCurrentState());
+ currentStateDelta.setState(
+ partitionKey,
+ (stateModel.getCurrentState() == null) ? State.from(initState) : State.from(stateModel
+ .getCurrentState()));
return new HelixStateTransitionHandler(stateModelFactory, stateModel, message, context,
currentStateDelta);
} else {
- BatchMessageWrapper wrapper = stateModelFactory.getBatchMessageWrapper(resourceName);
+ BatchMessageWrapper wrapper =
+ stateModelFactory.getBatchMessageWrapper(resourceId.stringify());
if (wrapper == null) {
- wrapper = stateModelFactory.createAndAddBatchMessageWrapper(resourceName);
+ wrapper = stateModelFactory.createAndAddBatchMessageWrapper(resourceId.stringify());
}
// get executor-service for the message
TaskExecutor executor = (TaskExecutor) context.get(MapKey.TASK_EXECUTOR.toString());
if (executor == null) {
logger.error("fail to get executor-service for batch message: " + message.getId()
- + ". msgType: " + message.getMsgType() + ", resource: " + message.getResourceName());
+ + ". msgType: " + message.getMsgType() + ", resource: " + message.getResourceId());
return null;
}
return new BatchMessageHandler(message, context, this, wrapper, executor);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/tools/MessagePoster.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/MessagePoster.java b/helix-core/src/main/java/org/apache/helix/tools/MessagePoster.java
index f0114d0..57d9828 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/MessagePoster.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/MessagePoster.java
@@ -22,10 +22,13 @@ package org.apache.helix.tools;
import java.util.UUID;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.State;
import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.manager.zk.ZkClient;
-import org.apache.helix.model.Message;
import org.apache.helix.model.LiveInstance.LiveInstanceProperty;
+import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageState;
import org.apache.helix.model.Message.MessageType;
import org.apache.helix.util.HelixUtil;
@@ -37,8 +40,8 @@ public class MessagePoster {
String path = HelixUtil.getMessagePath(clusterName, instanceName) + "/" + message.getId();
client.delete(path);
ZNRecord record = client.readData(HelixUtil.getLiveInstancePath(clusterName, instanceName));
- message.setTgtSessionId(record.getSimpleField(LiveInstanceProperty.SESSION_ID.toString())
- .toString());
+ message.setTgtSessionId(Id.session(record.getSimpleField(
+ LiveInstanceProperty.SESSION_ID.toString()).toString()));
message.setTgtName(record.getId());
// System.out.println(message);
client.createPersistent(path, message.getRecord());
@@ -46,12 +49,12 @@ public class MessagePoster {
public void postFaultInjectionMessage(String zkServer, String clusterName, String instanceName,
String payloadString, String partition) {
- Message message = new Message("FaultInjection", UUID.randomUUID().toString());
+ Message message = new Message("FaultInjection", Id.message(UUID.randomUUID().toString()));
if (payloadString != null) {
message.getRecord().setSimpleField("faultType", payloadString);
}
if (partition != null) {
- message.setPartitionName(partition);
+ message.setPartitionId(Id.partition(partition));
}
post(zkServer, message, clusterName, instanceName);
@@ -59,16 +62,16 @@ public class MessagePoster {
public void postTestMessage(String zkServer, String clusterName, String instanceName) {
String msgSrc = "cm-instance-0";
- String msgId = "TestMessageId-2";
+ MessageId msgId = Id.message("TestMessageId-2");
Message message = new Message(MessageType.STATE_TRANSITION, msgId);
message.setMsgId(msgId);
message.setSrcName(msgSrc);
message.setTgtName(instanceName);
message.setMsgState(MessageState.NEW);
- message.setFromState("Slave");
- message.setToState("Master");
- message.setPartitionName("EspressoDB.partition-0." + instanceName);
+ message.setFromState(State.from("Slave"));
+ message.setToState(State.from("Master"));
+ message.setPartitionId(Id.partition("EspressoDB.partition-0." + instanceName));
post(zkServer, message, clusterName, instanceName);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/tools/StateModelConfigGenerator.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/StateModelConfigGenerator.java b/helix-core/src/main/java/org/apache/helix/tools/StateModelConfigGenerator.java
index 508e447..353ba81 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/StateModelConfigGenerator.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/StateModelConfigGenerator.java
@@ -27,8 +27,8 @@ import java.util.Map;
import org.apache.helix.ZNRecord;
import org.apache.helix.manager.zk.DefaultSchedulerMessageHandlerFactory;
import org.apache.helix.manager.zk.ZNRecordSerializer;
-import org.apache.helix.model.Transition;
import org.apache.helix.model.StateModelDefinition.StateModelDefinitionProperty;
+import org.apache.helix.model.Transition;
import org.apache.helix.model.builder.StateTransitionTableBuilder;
// TODO refactor to use StateModelDefinition.Builder
@@ -328,9 +328,9 @@ public class StateModelConfigGenerator {
states.add("OFFLINE");
List<Transition> transitions = new ArrayList<Transition>();
- transitions.add(new Transition("OFFLINE", "COMPLETED"));
- transitions.add(new Transition("OFFLINE", "DROPPED"));
- transitions.add(new Transition("COMPLETED", "DROPPED"));
+ transitions.add(Transition.from("OFFLINE-COMPLETED"));
+ transitions.add(Transition.from("OFFLINE-DROPPED"));
+ transitions.add(Transition.from("COMPLETED-DROPPED"));
StateTransitionTableBuilder builder = new StateTransitionTableBuilder();
Map<String, Map<String, String>> next = builder.buildTransitionTable(states, transitions);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/tools/ZkLogAnalyzer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ZkLogAnalyzer.java b/helix-core/src/main/java/org/apache/helix/tools/ZkLogAnalyzer.java
index 1cf824c..58186d5 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ZkLogAnalyzer.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ZkLogAnalyzer.java
@@ -342,11 +342,14 @@ public class ZkLogAnalyzer {
// sendMessageLines.add(inputLine);
stats.msgSentCount++;
- if (msg.getFromStateString().equals("OFFLINE") && msg.getToStateString().equals("SLAVE")) {
+ if (msg.getFromState().toString().equals("OFFLINE")
+ && msg.getToState().toString().equals("SLAVE")) {
stats.msgSentCount_O2S++;
- } else if (msg.getFromStateString().equals("SLAVE") && msg.getToStateString().equals("MASTER")) {
+ } else if (msg.getFromState().toString().equals("SLAVE")
+ && msg.getToState().toString().equals("MASTER")) {
stats.msgSentCount_S2M++;
- } else if (msg.getFromStateString().equals("MASTER") && msg.getToStateString().equals("SLAVE")) {
+ } else if (msg.getFromState().toString().equals("MASTER")
+ && msg.getToState().toString().equals("SLAVE")) {
stats.msgSentCount_M2S++;
}
// System.out.println("Message create:"+new
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java b/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
index 7fab9d5..37d4568 100644
--- a/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
@@ -38,6 +38,8 @@ import org.apache.helix.HelixProperty;
import org.apache.helix.PropertyKey;
import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.SessionId;
import org.apache.helix.model.Error;
import org.apache.helix.model.Message;
import org.apache.helix.model.StatusUpdate;
@@ -254,7 +256,7 @@ public class StatusUpdateUtil {
*/
ZNRecord createMessageLogRecord(Message message) {
ZNRecord result = new ZNRecord(getStatusUpdateRecordName(message));
- String mapFieldKey = "MESSAGE " + message.getMsgIdString();
+ String mapFieldKey = "MESSAGE " + message.getMsgId();
result.setMapField(mapFieldKey, new TreeMap<String, String>());
// Store all the simple fields of the message in the new ZNRecord's map
@@ -290,7 +292,7 @@ public class StatusUpdateUtil {
contentMap.put("Message state", message.getMsgState().toString());
contentMap.put("AdditionalInfo", additionalInfo);
contentMap.put("Class", classInfo.toString());
- contentMap.put("MSG_ID", message.getMsgIdString());
+ contentMap.put("MSG_ID", message.getMsgId().stringify());
DateFormat formatter = new SimpleDateFormat("yyyyMMdd-HHmmss.SSSSSS");
String time = formatter.format(new Date());
@@ -304,8 +306,8 @@ public class StatusUpdateUtil {
String getRecordIdForMessage(Message message) {
if (message.getMsgType().equals(MessageType.STATE_TRANSITION)) {
- return message.getPartitionName() + " Trans:" + message.getFromStateString().charAt(0) + "->"
- + message.getToStateString().charAt(0) + " " + UUID.randomUUID().toString();
+ return message.getPartitionId() + " Trans:" + message.getFromState().toString().charAt(0)
+ + "->" + message.getToState().toString().charAt(0) + " " + UUID.randomUUID().toString();
} else {
return message.getMsgType() + " " + UUID.randomUUID().toString();
}
@@ -375,16 +377,16 @@ public class StatusUpdateUtil {
String instanceName = message.getTgtName();
String statusUpdateSubPath = getStatusUpdateSubPath(message);
String statusUpdateKey = getStatusUpdateKey(message);
- String sessionId = message.getExecutionSessionIdString();
+ SessionId sessionId = message.getExecutionSessionId();
if (sessionId == null) {
- sessionId = message.getTgtSessionIdString();
+ sessionId = message.getTgtSessionId();
}
if (sessionId == null) {
- sessionId = "*";
+ sessionId = Id.session("*");
}
Builder keyBuilder = accessor.keyBuilder();
- if (!_recordedMessages.containsKey(message.getMsgIdString())) {
+ if (!_recordedMessages.containsKey(message.getMsgId().stringify())) {
// TODO instanceName of a controller might be any string
if (instanceName.equalsIgnoreCase("Controller")) {
accessor.updateProperty(
@@ -394,8 +396,8 @@ public class StatusUpdateUtil {
} else {
PropertyKey propertyKey =
- keyBuilder.stateTransitionStatus(instanceName, sessionId, statusUpdateSubPath,
- statusUpdateKey);
+ keyBuilder.stateTransitionStatus(instanceName, sessionId.stringify(),
+ statusUpdateSubPath, statusUpdateKey);
ZNRecord statusUpdateRecord = createMessageLogRecord(message);
@@ -408,7 +410,7 @@ public class StatusUpdateUtil {
accessor.updateProperty(propertyKey, new StatusUpdate(statusUpdateRecord));
}
- _recordedMessages.put(message.getMsgIdString(), message.getMsgIdString());
+ _recordedMessages.put(message.getMsgId().stringify(), message.getMsgId().stringify());
}
if (instanceName.equalsIgnoreCase("Controller")) {
@@ -418,8 +420,8 @@ public class StatusUpdateUtil {
} else {
PropertyKey propertyKey =
- keyBuilder.stateTransitionStatus(instanceName, sessionId, statusUpdateSubPath,
- statusUpdateKey);
+ keyBuilder.stateTransitionStatus(instanceName, sessionId.stringify(),
+ statusUpdateSubPath, statusUpdateKey);
// For now write participant StatusUpdates to log4j.
// we are using restlet as another data channel to report to controller.
if (_logger.isTraceEnabled()) {
@@ -436,9 +438,9 @@ public class StatusUpdateUtil {
private String getStatusUpdateKey(Message message) {
if (message.getMsgType().equalsIgnoreCase(MessageType.STATE_TRANSITION.toString())) {
- return message.getPartitionName();
+ return message.getPartitionId().stringify();
}
- return message.getMsgIdString();
+ return message.getMsgId().stringify();
}
/**
@@ -446,7 +448,7 @@ public class StatusUpdateUtil {
*/
String getStatusUpdateSubPath(Message message) {
if (message.getMsgType().equalsIgnoreCase(MessageType.STATE_TRANSITION.toString())) {
- return message.getResourceName();
+ return message.getResourceId().stringify();
} else {
return message.getMsgType();
}
@@ -454,9 +456,9 @@ public class StatusUpdateUtil {
String getStatusUpdateRecordName(Message message) {
if (message.getMsgType().equalsIgnoreCase(MessageType.STATE_TRANSITION.toString())) {
- return message.getTgtSessionIdString() + "__" + message.getResourceName();
+ return message.getTgtSessionId() + "__" + message.getResourceId();
}
- return message.getMsgIdString();
+ return message.getMsgId().stringify();
}
/**
@@ -472,12 +474,12 @@ public class StatusUpdateUtil {
String instanceName = message.getTgtName();
String statusUpdateSubPath = getStatusUpdateSubPath(message);
String statusUpdateKey = getStatusUpdateKey(message);
- String sessionId = message.getExecutionSessionIdString();
+ SessionId sessionId = message.getExecutionSessionId();
if (sessionId == null) {
- sessionId = message.getTgtSessionIdString();
+ sessionId = message.getTgtSessionId();
}
if (sessionId == null) {
- sessionId = "*";
+ sessionId = Id.session("*");
}
Builder keyBuilder = accessor.keyBuilder();
@@ -496,7 +498,7 @@ public class StatusUpdateUtil {
// sessionId,
// statusUpdateSubPath,
// statusUpdateKey);
- accessor.updateProperty(keyBuilder.stateTransitionError(instanceName, sessionId,
+ accessor.updateProperty(keyBuilder.stateTransitionError(instanceName, sessionId.stringify(),
statusUpdateSubPath, statusUpdateKey), new Error(record));
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/TestHelixTaskExecutor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/TestHelixTaskExecutor.java b/helix-core/src/test/java/org/apache/helix/TestHelixTaskExecutor.java
index 4aa16eb..4cf0744 100644
--- a/helix-core/src/test/java/org/apache/helix/TestHelixTaskExecutor.java
+++ b/helix-core/src/test/java/org/apache/helix/TestHelixTaskExecutor.java
@@ -23,6 +23,9 @@ import org.apache.helix.Mocks.MockHelixTaskExecutor;
import org.apache.helix.Mocks.MockManager;
import org.apache.helix.Mocks.MockStateModel;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.State;
import org.apache.helix.messaging.handling.AsyncCallbackService;
import org.apache.helix.messaging.handling.HelixStateTransitionHandler;
import org.apache.helix.messaging.handling.HelixTask;
@@ -40,18 +43,18 @@ public class TestHelixTaskExecutor {
@Test()
public void testCMTaskExecutor() throws Exception {
System.out.println("START TestCMTaskExecutor");
- String msgId = "TestMessageId";
+ MessageId msgId = Id.message("TestMessageId");
Message message = new Message(MessageType.TASK_REPLY, msgId);
message.setMsgId(msgId);
message.setSrcName("cm-instance-0");
message.setTgtName("cm-instance-1");
- message.setTgtSessionId("1234");
- message.setFromState("Offline");
- message.setToState("Slave");
- message.setPartitionName("TestDB_0");
- message.setResourceName("TestDB");
- message.setStateModelDef("MasterSlave");
+ message.setTgtSessionId(Id.session("1234"));
+ message.setFromState(State.from("Offline"));
+ message.setToState(State.from("Slave"));
+ message.setPartitionId(Id.partition("TestDB_0"));
+ message.setResourceId(Id.resource("TestDB"));
+ message.setStateModelDef(Id.stateModelDef("MasterSlave"));
MockManager manager = new MockManager("clusterName");
HelixDataAccessor accessor = manager.getHelixDataAccessor();
@@ -67,7 +70,7 @@ public class TestHelixTaskExecutor {
NotificationContext context = new NotificationContext(manager);
CurrentState currentStateDelta = new CurrentState("TestDB");
- currentStateDelta.setState("TestDB_0", "OFFLINE");
+ currentStateDelta.setState(Id.partition("TestDB_0"), State.from("OFFLINE"));
StateModelFactory<MockStateModel> stateModelFactory = new StateModelFactory<MockStateModel>() {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/TestHelixTaskHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/TestHelixTaskHandler.java b/helix-core/src/test/java/org/apache/helix/TestHelixTaskHandler.java
index 3c7f2af..3614425 100644
--- a/helix-core/src/test/java/org/apache/helix/TestHelixTaskHandler.java
+++ b/helix-core/src/test/java/org/apache/helix/TestHelixTaskHandler.java
@@ -21,20 +21,19 @@ package org.apache.helix;
import java.util.Date;
-import org.apache.helix.HelixConstants;
-import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.NotificationContext;
import org.apache.helix.Mocks.MockManager;
import org.apache.helix.Mocks.MockStateModel;
import org.apache.helix.Mocks.MockStateModelAnnotated;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.State;
import org.apache.helix.messaging.handling.HelixStateTransitionHandler;
import org.apache.helix.messaging.handling.HelixTask;
import org.apache.helix.messaging.handling.HelixTaskExecutor;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
-import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.participant.statemachine.StateModelFactory;
import org.apache.helix.tools.StateModelConfigGenerator;
import org.testng.AssertJUnit;
@@ -45,17 +44,17 @@ public class TestHelixTaskHandler {
public void testInvocation() throws Exception {
HelixTaskExecutor executor = new HelixTaskExecutor();
System.out.println("START TestCMTaskHandler.testInvocation()");
- Message message = new Message(MessageType.STATE_TRANSITION, "Some unique id");
+ Message message = new Message(MessageType.STATE_TRANSITION, Id.message("Some unique id"));
message.setSrcName("cm-instance-0");
- message.setTgtSessionId("1234");
- message.setFromState("Offline");
- message.setToState("Slave");
- message.setPartitionName("TestDB_0");
- message.setMsgId("Some unique message id");
- message.setResourceName("TestDB");
+ message.setTgtSessionId(Id.session("1234"));
+ message.setFromState(State.from("Offline"));
+ message.setToState(State.from("Slave"));
+ message.setPartitionId(Id.partition("TestDB_0"));
+ message.setMsgId(Id.message("Some unique message id"));
+ message.setResourceId(Id.resource("TestDB"));
message.setTgtName("localhost");
- message.setStateModelDef("MasterSlave");
+ message.setStateModelDef(Id.stateModelDef("MasterSlave"));
message.setStateModelFactoryName(HelixConstants.DEFAULT_STATE_MODEL_FACTORY);
MockStateModel stateModel = new MockStateModel();
NotificationContext context;
@@ -68,7 +67,7 @@ public class TestHelixTaskHandler {
context = new NotificationContext(manager);
CurrentState currentStateDelta = new CurrentState("TestDB");
- currentStateDelta.setState("TestDB_0", "OFFLINE");
+ currentStateDelta.setState(Id.partition("TestDB_0"), State.from("OFFLINE"));
HelixStateTransitionHandler stHandler =
new HelixStateTransitionHandler(null, stateModel, message, context, currentStateDelta);
@@ -85,16 +84,16 @@ public class TestHelixTaskHandler {
System.out.println("START TestCMTaskHandler.testInvocationAnnotated() at "
+ new Date(System.currentTimeMillis()));
HelixTaskExecutor executor = new HelixTaskExecutor();
- Message message = new Message(MessageType.STATE_TRANSITION, "Some unique id");
+ Message message = new Message(MessageType.STATE_TRANSITION, Id.message("Some unique id"));
message.setSrcName("cm-instance-0");
- message.setTgtSessionId("1234");
- message.setFromState("Offline");
- message.setToState("Slave");
- message.setPartitionName("TestDB_0");
- message.setMsgId("Some unique message id");
- message.setResourceName("TestDB");
+ message.setTgtSessionId(Id.session("1234"));
+ message.setFromState(State.from("Offline"));
+ message.setToState(State.from("Slave"));
+ message.setPartitionId(Id.partition("TestDB_0"));
+ message.setMsgId(Id.message("Some unique message id"));
+ message.setResourceId(Id.resource("TestDB"));
message.setTgtName("localhost");
- message.setStateModelDef("MasterSlave");
+ message.setStateModelDef(Id.stateModelDef("MasterSlave"));
message.setStateModelFactoryName(HelixConstants.DEFAULT_STATE_MODEL_FACTORY);
MockStateModelAnnotated stateModel = new MockStateModelAnnotated();
NotificationContext context;
@@ -110,7 +109,7 @@ public class TestHelixTaskHandler {
context = new NotificationContext(manager);
CurrentState currentStateDelta = new CurrentState("TestDB");
- currentStateDelta.setState("TestDB_0", "OFFLINE");
+ currentStateDelta.setState(Id.partition("TestDB_0"), State.from("OFFLINE"));
StateModelFactory<MockStateModelAnnotated> stateModelFactory =
new StateModelFactory<MockStateModelAnnotated>() {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/TestHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/TestHelper.java b/helix-core/src/test/java/org/apache/helix/TestHelper.java
index 5c4749e..edc9161 100644
--- a/helix-core/src/test/java/org/apache/helix/TestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/TestHelper.java
@@ -45,6 +45,9 @@ import org.I0Itec.zkclient.ZkServer;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.apache.commons.io.FileUtils;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.integration.manager.ZkTestManager;
import org.apache.helix.manager.zk.CallbackHandler;
@@ -494,15 +497,15 @@ public class TestHelper {
return resultsMap;
}
- public static Message createMessage(String msgId, String fromState, String toState,
+ public static Message createMessage(MessageId msgId, String fromState, String toState,
String tgtName, String resourceName, String partitionName) {
Message msg = new Message(MessageType.STATE_TRANSITION, msgId);
- msg.setFromState(fromState);
- msg.setToState(toState);
+ msg.setFromState(State.from(fromState));
+ msg.setToState(State.from(toState));
msg.setTgtName(tgtName);
- msg.setResourceName(resourceName);
- msg.setPartitionName(partitionName);
- msg.setStateModelDef("MasterSlave");
+ msg.setResourceId(Id.resource(resourceName));
+ msg.setPartitionId(Id.partition(partitionName));
+ msg.setStateModelDef(Id.stateModelDef("MasterSlave"));
return msg;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/TestZKCallback.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/TestZKCallback.java b/helix-core/src/test/java/org/apache/helix/TestZKCallback.java
index 72799f0..8ae02c7 100644
--- a/helix-core/src/test/java/org/apache/helix/TestZKCallback.java
+++ b/helix-core/src/test/java/org/apache/helix/TestZKCallback.java
@@ -37,6 +37,8 @@ import org.apache.helix.LiveInstanceChangeListener;
import org.apache.helix.MessageListener;
import org.apache.helix.NotificationContext;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.State;
import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.CurrentState;
@@ -155,7 +157,7 @@ public class TestZKCallback extends ZkUnitTestBase {
testListener.Reset();
CurrentState curState = new CurrentState("db-12345");
- curState.setSessionId("sessionId");
+ curState.setSessionId(Id.session("sessionId"));
curState.setStateModelDefRef("StateModelDef");
accessor.setProperty(keyBuilder.currentState("localhost_8900", testHelixManager.getSessionId(),
curState.getId()), curState);
@@ -184,13 +186,14 @@ public class TestZKCallback extends ZkUnitTestBase {
// recList.add(dummyRecord);
testListener.Reset();
- Message message = new Message(MessageType.STATE_TRANSITION, UUID.randomUUID().toString());
- message.setTgtSessionId("*");
- message.setResourceName("testResource");
- message.setPartitionName("testPartitionKey");
- message.setStateModelDef("MasterSlave");
- message.setToState("toState");
- message.setFromState("fromState");
+ Message message =
+ new Message(MessageType.STATE_TRANSITION, Id.message(UUID.randomUUID().toString()));
+ message.setTgtSessionId(Id.session("*"));
+ message.setResourceId(Id.resource("testResource"));
+ message.setPartitionId(Id.partition("testPartitionKey"));
+ message.setStateModelDef(Id.stateModelDef("MasterSlave"));
+ message.setToState(State.from("toState"));
+ message.setFromState(State.from("fromState"));
message.setTgtName("testTarget");
message.setStateModelFactoryName(HelixConstants.DEFAULT_STATE_MODEL_FACTORY);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/ZkUnitTestBase.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/ZkUnitTestBase.java b/helix-core/src/test/java/org/apache/helix/ZkUnitTestBase.java
index 4a505dc..2425020 100644
--- a/helix-core/src/test/java/org/apache/helix/ZkUnitTestBase.java
+++ b/helix-core/src/test/java/org/apache/helix/ZkUnitTestBase.java
@@ -28,6 +28,8 @@ import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkConnection;
import org.I0Itec.zkclient.ZkServer;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.pipeline.Pipeline;
import org.apache.helix.controller.pipeline.Stage;
import org.apache.helix.controller.pipeline.StageContext;
@@ -344,11 +346,11 @@ public class ZkUnitTestBase {
stage.postProcess();
}
- protected Message createMessage(MessageType type, String msgId, String fromState, String toState,
- String resourceName, String tgtName) {
+ protected Message createMessage(MessageType type, MessageId msgId, String fromState,
+ String toState, String resourceName, String tgtName) {
Message msg = new Message(type.toString(), msgId);
- msg.setFromState(fromState);
- msg.setToState(toState);
+ msg.setFromState(State.from(fromState));
+ msg.setToState(State.from(toState));
msg.getRecord().setSimpleField(Attributes.RESOURCE_NAME.toString(), resourceName);
msg.setTgtName(tgtName);
return msg;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/controller/stages/TestCurrentStateComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestCurrentStateComputationStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestCurrentStateComputationStage.java
index 7687e18..ecad444 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestCurrentStateComputationStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestCurrentStateComputationStage.java
@@ -21,12 +21,10 @@ package org.apache.helix.controller.stages;
import java.util.Map;
-import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
-import org.apache.helix.controller.stages.AttributeName;
-import org.apache.helix.controller.stages.CurrentStateComputationStage;
-import org.apache.helix.controller.stages.CurrentStateOutput;
-import org.apache.helix.controller.stages.ReadClusterDataStage;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.State;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Partition;
@@ -66,13 +64,13 @@ public class TestCurrentStateComputationStage extends BaseStageTest {
0);
// Add a state transition messages
- Message message = new Message(Message.MessageType.STATE_TRANSITION, "msg1");
- message.setFromState("OFFLINE");
- message.setToState("SLAVE");
- message.setResourceName("testResourceName");
- message.setPartitionName("testResourceName_1");
+ Message message = new Message(Message.MessageType.STATE_TRANSITION, Id.message("msg1"));
+ message.setFromState(State.from("OFFLINE"));
+ message.setToState(State.from("SLAVE"));
+ message.setResourceId(Id.resource("testResourceName"));
+ message.setPartitionId(Id.partition("testResourceName_1"));
message.setTgtName("localhost_3");
- message.setTgtSessionId("session_3");
+ message.setTgtSessionId(Id.session("session_3"));
Builder keyBuilder = accessor.keyBuilder();
accessor.setProperty(keyBuilder.message("localhost_" + 3, message.getId()), message);
@@ -88,14 +86,14 @@ public class TestCurrentStateComputationStage extends BaseStageTest {
ZNRecord record1 = new ZNRecord("testResourceName");
// Add a current state that matches sessionId and one that does not match
CurrentState stateWithLiveSession = new CurrentState(record1);
- stateWithLiveSession.setSessionId("session_3");
+ stateWithLiveSession.setSessionId(Id.session("session_3"));
stateWithLiveSession.setStateModelDefRef("MasterSlave");
- stateWithLiveSession.setState("testResourceName_1", "OFFLINE");
+ stateWithLiveSession.setState(Id.partition("testResourceName_1"), State.from("OFFLINE"));
ZNRecord record2 = new ZNRecord("testResourceName");
CurrentState stateWithDeadSession = new CurrentState(record2);
- stateWithDeadSession.setSessionId("session_dead");
+ stateWithDeadSession.setSessionId(Id.session("session_dead"));
stateWithDeadSession.setStateModelDefRef("MasterSlave");
- stateWithDeadSession.setState("testResourceName_1", "MASTER");
+ stateWithDeadSession.setState(Id.partition("testResourceName_1"), State.from("MASTER"));
accessor.setProperty(keyBuilder.currentState("localhost_3", "session_3", "testResourceName"),
stateWithLiveSession);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/controller/stages/TestMessageThrottleStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestMessageThrottleStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestMessageThrottleStage.java
index cdf11fe..26bbc20 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestMessageThrottleStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestMessageThrottleStage.java
@@ -28,26 +28,20 @@ import java.util.TreeMap;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.ZNRecord;
import org.apache.helix.ZkUnitTestBase;
-import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
import org.apache.helix.controller.pipeline.Pipeline;
-import org.apache.helix.controller.stages.AttributeName;
-import org.apache.helix.controller.stages.ClusterEvent;
-import org.apache.helix.controller.stages.MessageSelectionStageOutput;
-import org.apache.helix.controller.stages.MessageThrottleStage;
-import org.apache.helix.controller.stages.MessageThrottleStageOutput;
-import org.apache.helix.controller.stages.ReadClusterDataStage;
-import org.apache.helix.controller.stages.ResourceComputationStage;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.ClusterConstraints;
-import org.apache.helix.model.ConstraintItem;
-import org.apache.helix.model.Message;
-import org.apache.helix.model.Partition;
import org.apache.helix.model.ClusterConstraints.ConstraintAttribute;
import org.apache.helix.model.ClusterConstraints.ConstraintType;
+import org.apache.helix.model.ConstraintItem;
+import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.Partition;
import org.apache.log4j.Logger;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -109,8 +103,8 @@ public class TestMessageThrottleStage extends ZkUnitTestBase {
MessageSelectionStageOutput msgSelectOutput = new MessageSelectionStageOutput();
List<Message> selectMessages = new ArrayList<Message>();
Message msg =
- createMessage(MessageType.STATE_TRANSITION, "msgId-001", "OFFLINE", "SLAVE", "TestDB",
- "localhost_0");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-001"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_0");
selectMessages.add(msg);
msgSelectOutput.addMessages("TestDB", new Partition("TestDB_0"), selectMessages);
@@ -223,8 +217,8 @@ public class TestMessageThrottleStage extends ZkUnitTestBase {
// test constraintSelection
// message1: hit contraintSelection rule1 and rule2
Message msg1 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-001", "OFFLINE", "SLAVE", "TestDB",
- "localhost_0");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-001"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_0");
Map<ConstraintAttribute, String> msgAttr = ClusterConstraints.toConstraintAttributes(msg1);
Set<ConstraintItem> matches = constraint.match(msgAttr);
@@ -244,8 +238,8 @@ public class TestMessageThrottleStage extends ZkUnitTestBase {
// message2: hit contraintSelection rule1, rule2, and rule3
Message msg2 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-002", "OFFLINE", "SLAVE", "TestDB",
- "localhost_1");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-002"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_1");
msgAttr = ClusterConstraints.toConstraintAttributes(msg2);
matches = constraint.match(msgAttr);
@@ -274,20 +268,20 @@ public class TestMessageThrottleStage extends ZkUnitTestBase {
MessageSelectionStageOutput msgSelectOutput = new MessageSelectionStageOutput();
Message msg3 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-003", "OFFLINE", "SLAVE", "TestDB",
- "localhost_0");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-003"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_0");
Message msg4 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-004", "OFFLINE", "SLAVE", "TestDB",
- "localhost_0");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-004"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_0");
Message msg5 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-005", "OFFLINE", "SLAVE", "TestDB",
- "localhost_0");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-005"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_0");
Message msg6 =
- createMessage(MessageType.STATE_TRANSITION, "msgId-006", "OFFLINE", "SLAVE", "TestDB",
- "localhost_1");
+ createMessage(MessageType.STATE_TRANSITION, Id.message("msgId-006"), "OFFLINE", "SLAVE",
+ "TestDB", "localhost_1");
List<Message> selectMessages = new ArrayList<Message>();
selectMessages.add(msg1);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/controller/stages/TestMsgSelectionStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestMsgSelectionStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestMsgSelectionStage.java
index b8166f6..97d5ec1 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestMsgSelectionStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestMsgSelectionStage.java
@@ -26,7 +26,7 @@ import java.util.List;
import java.util.Map;
import org.apache.helix.TestHelper;
-import org.apache.helix.controller.stages.MessageSelectionStage;
+import org.apache.helix.api.Id;
import org.apache.helix.controller.stages.MessageSelectionStage.Bounds;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
@@ -49,10 +49,10 @@ public class TestMsgSelectionStage {
Map<String, String> pendingStates = new HashMap<String, String>();
List<Message> messages = new ArrayList<Message>();
- messages.add(TestHelper.createMessage("msgId_0", "SLAVE", "MASTER", "localhost_0", "TestDB",
- "TestDB_0"));
- messages.add(TestHelper.createMessage("msgId_1", "MASTER", "SLAVE", "localhost_1", "TestDB",
- "TestDB_0"));
+ messages.add(TestHelper.createMessage(Id.message("msgId_0"), "SLAVE", "MASTER", "localhost_0",
+ "TestDB", "TestDB_0"));
+ messages.add(TestHelper.createMessage(Id.message("msgId_1"), "MASTER", "SLAVE", "localhost_1",
+ "TestDB", "TestDB_0"));
Map<String, Bounds> stateConstraints = new HashMap<String, Bounds>();
stateConstraints.put("MASTER", new Bounds(0, 1));
@@ -67,7 +67,7 @@ public class TestMsgSelectionStage {
messages, stateConstraints, stateTransitionPriorities, "OFFLINE");
Assert.assertEquals(selectedMsg.size(), 1);
- Assert.assertEquals(selectedMsg.get(0).getMsgIdString(), "msgId_1");
+ Assert.assertEquals(selectedMsg.get(0).getMsgId(), Id.message("msgId_1"));
System.out.println("END testMasterXfer at " + new Date(System.currentTimeMillis()));
}
@@ -88,8 +88,8 @@ public class TestMsgSelectionStage {
pendingStates.put("localhost_1", "MASTER");
List<Message> messages = new ArrayList<Message>();
- messages.add(TestHelper.createMessage("msgId_0", "SLAVE", "MASTER", "localhost_0", "TestDB",
- "TestDB_0"));
+ messages.add(TestHelper.createMessage(Id.message("msgId_0"), "SLAVE", "MASTER", "localhost_0",
+ "TestDB", "TestDB_0"));
Map<String, Bounds> stateConstraints = new HashMap<String, Bounds>();
stateConstraints.put("MASTER", new Bounds(0, 1));
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
index 4c5a0d7..33dce89 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
@@ -25,30 +25,21 @@ import java.util.List;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.TestHelper;
import org.apache.helix.ZNRecord;
import org.apache.helix.ZkUnitTestBase;
-import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.State;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.controller.pipeline.Pipeline;
-import org.apache.helix.controller.stages.AttributeName;
-import org.apache.helix.controller.stages.BestPossibleStateCalcStage;
-import org.apache.helix.controller.stages.ClusterEvent;
-import org.apache.helix.controller.stages.CurrentStateComputationStage;
-import org.apache.helix.controller.stages.MessageGenerationPhase;
-import org.apache.helix.controller.stages.MessageSelectionStage;
-import org.apache.helix.controller.stages.MessageSelectionStageOutput;
-import org.apache.helix.controller.stages.MessageThrottleStage;
-import org.apache.helix.controller.stages.ReadClusterDataStage;
-import org.apache.helix.controller.stages.ResourceComputationStage;
-import org.apache.helix.controller.stages.TaskAssignmentStage;
import org.apache.helix.manager.zk.ZKHelixAdmin;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
-import org.apache.helix.model.Partition;
import org.apache.helix.model.Message.Attributes;
+import org.apache.helix.model.Partition;
import org.apache.log4j.Logger;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -111,8 +102,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
msgSelOutput.getMessages(resourceName, new Partition(resourceName + "_0"));
Assert.assertEquals(messages.size(), 1, "Should output 1 message: OFFLINE-SLAVE for node0");
Message message = messages.get(0);
- Assert.assertEquals(message.getFromStateString(), "OFFLINE");
- Assert.assertEquals(message.getToStateString(), "SLAVE");
+ Assert.assertEquals(message.getFromState().toString(), "OFFLINE");
+ Assert.assertEquals(message.getToState().toString(), "SLAVE");
Assert.assertEquals(message.getTgtName(), "localhost_0");
// round2: updates node0 currentState to SLAVE but keep the
@@ -258,8 +249,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
msgSelOutput.getMessages(resourceName, new Partition(resourceName + "_0"));
Assert.assertEquals(messages.size(), 1, "Should output 1 message: OFFLINE-SLAVE for node0");
Message message = messages.get(0);
- Assert.assertEquals(message.getFromStateString(), "OFFLINE");
- Assert.assertEquals(message.getToStateString(), "SLAVE");
+ Assert.assertEquals(message.getFromState().toString(), "OFFLINE");
+ Assert.assertEquals(message.getToState().toString(), "SLAVE");
Assert.assertEquals(message.getTgtName(), "localhost_0");
// round2: drop resource, but keep the
@@ -275,8 +266,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
"Should output only 1 message: OFFLINE->DROPPED for localhost_1");
message = messages.get(0);
- Assert.assertEquals(message.getFromStateString(), "SLAVE");
- Assert.assertEquals(message.getToStateString(), "OFFLINE");
+ Assert.assertEquals(message.getFromState().toString(), "SLAVE");
+ Assert.assertEquals(message.getToState().toString(), "OFFLINE");
Assert.assertEquals(message.getTgtName(), "localhost_1");
// round3: remove O->S for localhost_0, controller should now send O->DROPPED to
@@ -291,8 +282,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
Assert.assertEquals(messages.size(), 1,
"Should output 1 message: OFFLINE->DROPPED for localhost_0");
message = messages.get(0);
- Assert.assertEquals(message.getFromStateString(), "OFFLINE");
- Assert.assertEquals(message.getToStateString(), "DROPPED");
+ Assert.assertEquals(message.getFromState().toString(), "OFFLINE");
+ Assert.assertEquals(message.getToState().toString(), "DROPPED");
Assert.assertEquals(message.getTgtName(), "localhost_0");
System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
@@ -351,8 +342,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
msgSelOutput.getMessages(resourceName, new Partition(resourceName + "_0"));
Assert.assertEquals(messages.size(), 1, "Should output 1 message: SLAVE-MASTER for node1");
Message message = messages.get(0);
- Assert.assertEquals(message.getFromStateString(), "SLAVE");
- Assert.assertEquals(message.getToStateString(), "MASTER");
+ Assert.assertEquals(message.getFromState().toString(), "SLAVE");
+ Assert.assertEquals(message.getToState().toString(), "MASTER");
Assert.assertEquals(message.getTgtName(), "localhost_1");
// round2: updates node0 currentState to SLAVE but keep the
@@ -380,8 +371,8 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
Builder keyBuilder = accessor.keyBuilder();
CurrentState curState = new CurrentState(resourceGroupName);
- curState.setState(resourceKey, state);
- curState.setSessionId(sessionId);
+ curState.setState(Id.partition(resourceKey), State.from(state));
+ curState.setSessionId(Id.session(sessionId));
curState.setStateModelDefRef("MasterSlave");
accessor.setProperty(keyBuilder.currentState(instance, sessionId, resourceGroupName), curState);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/controller/stages/TestResourceComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestResourceComputationStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestResourceComputationStage.java
index b29663c..86bd060 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestResourceComputationStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestResourceComputationStage.java
@@ -27,6 +27,8 @@ import java.util.UUID;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.State;
import org.apache.helix.controller.pipeline.StageContext;
import org.apache.helix.controller.strategy.DefaultTwoStateStrategy;
import org.apache.helix.model.CurrentState;
@@ -142,9 +144,9 @@ public class TestResourceComputationStage extends BaseStageTest {
String oldResource = "testResourceOld";
CurrentState currentState = new CurrentState(oldResource);
- currentState.setState("testResourceOld_0", "OFFLINE");
- currentState.setState("testResourceOld_1", "SLAVE");
- currentState.setState("testResourceOld_2", "MASTER");
+ currentState.setState(Id.partition("testResourceOld_0"), State.from("OFFLINE"));
+ currentState.setState(Id.partition("testResourceOld_1"), State.from("SLAVE"));
+ currentState.setState(Id.partition("testResourceOld_2"), State.from("MASTER"));
currentState.setStateModelDefRef("MasterSlave");
accessor.setProperty(keyBuilder.currentState(instanceName, sessionId, oldResource),
currentState);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/healthcheck/TestAddDropAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestAddDropAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestAddDropAlert.java
index 3de9d4d..69c2df6 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestAddDropAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestAddDropAlert.java
@@ -30,6 +30,8 @@ import org.apache.helix.TestHelper;
import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.TestHelper.StartCMResult;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.healthcheck.HealthStatsAggregationTask;
import org.apache.helix.healthcheck.ParticipantHealthReportCollectorImpl;
@@ -75,12 +77,12 @@ public class TestAddDropAlert extends ZkIntegrationTestBase {
public void doTransition(Message message, NotificationContext context) {
HelixManager manager = context.getManager();
HelixDataAccessor accessor = manager.getHelixDataAccessor();
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
String instance = message.getTgtName();
- String partition = message.getPartitionName();
- if (fromState.equalsIgnoreCase("SLAVE") && toState.equalsIgnoreCase("MASTER")) {
+ if (fromState.toString().equalsIgnoreCase("SLAVE")
+ && toState.toString().equalsIgnoreCase("MASTER")) {
// add a stat and report to ZK
// perhaps should keep reporter per instance...
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
index 211f230..6b545ed 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
@@ -31,6 +31,8 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.TestHelper.StartCMResult;
import org.apache.helix.alerts.AlertValueAndStatus;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.healthcheck.HealthStatsAggregationTask;
import org.apache.helix.healthcheck.ParticipantHealthReportCollectorImpl;
@@ -77,12 +79,12 @@ public class TestExpandAlert extends ZkIntegrationTestBase {
public void doTransition(Message message, NotificationContext context) {
HelixManager manager = context.getManager();
HelixDataAccessor accessor = manager.getHelixDataAccessor();
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
String instance = message.getTgtName();
- String partition = message.getPartitionName();
- if (fromState.equalsIgnoreCase("SLAVE") && toState.equalsIgnoreCase("MASTER")) {
+ if (fromState.toString().equalsIgnoreCase("SLAVE")
+ && toState.toString().equalsIgnoreCase("MASTER")) {
// add a stat and report to ZK
// perhaps should keep reporter per instance...
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
index f5eb13c..d10ed53 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
@@ -31,6 +31,7 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.TestHelper.StartCMResult;
import org.apache.helix.alerts.AlertValueAndStatus;
+import org.apache.helix.api.State;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.healthcheck.HealthStatsAggregationTask;
import org.apache.helix.healthcheck.ParticipantHealthReportCollectorImpl;
@@ -82,12 +83,12 @@ public class TestSimpleAlert extends ZkIntegrationTestBase {
public void doTransition(Message message, NotificationContext context) {
HelixManager manager = context.getManager();
HelixDataAccessor accessor = manager.getHelixDataAccessor();
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
String instance = message.getTgtName();
- String partition = message.getPartitionName();
- if (fromState.equalsIgnoreCase("SLAVE") && toState.equalsIgnoreCase("MASTER")) {
+ if (fromState.toString().equalsIgnoreCase("SLAVE")
+ && toState.toString().equalsIgnoreCase("MASTER")) {
// add a stat and report to ZK
// perhaps should keep reporter per instance...