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:30 UTC
[4/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/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 020c99d..8286d1b 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
@@ -39,6 +39,9 @@ import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
import org.apache.helix.ZNRecord;
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.AsyncCallback;
import org.apache.helix.messaging.handling.HelixTaskResult;
import org.apache.helix.messaging.handling.MessageHandler;
@@ -76,7 +79,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
@Override
public void onTimeOut() {
- _logger.info("Scheduler msg timeout " + _originalMessage.getMsgIdString() + " timout with "
+ _logger.info("Scheduler msg timeout " + _originalMessage.getMsgId() + " timout with "
+ _timeout + " Ms");
_statusUpdateUtil.logError(_originalMessage, SchedulerAsyncCallback.class, "Task timeout",
@@ -86,13 +89,13 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
@Override
public void onReplyMessage(Message message) {
- _logger.info("Update for scheduler msg " + _originalMessage.getMsgIdString() + " Message "
+ _logger.info("Update for scheduler msg " + _originalMessage.getMsgId() + " 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.getMsgIdString() + " completed");
+ _logger.info("Scheduler msg " + _originalMessage.getMsgId() + " completed");
_statusUpdateUtil.logInfo(_originalMessage, SchedulerAsyncCallback.class,
"Scheduler task completed", _manager.getHelixDataAccessor());
addSummary(_resultSummaryMap, _originalMessage, _manager, false);
@@ -110,13 +113,12 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
Builder keyBuilder = accessor.keyBuilder();
ZNRecord statusUpdate =
accessor.getProperty(
- keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- originalMessage.getMsgIdString())).getRecord();
+ keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(), originalMessage
+ .getMsgId().stringify())).getRecord();
statusUpdate.getMapFields().putAll(_resultSummaryMap);
- accessor.setProperty(
- keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- originalMessage.getMsgIdString()), new StatusUpdate(statusUpdate));
+ accessor.setProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
+ originalMessage.getMsgId().stringify()), new StatusUpdate(statusUpdate));
}
}
@@ -133,7 +135,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ message.getMsgType());
}
@@ -159,7 +161,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
}
void handleMessageUsingScheduledTaskQueue(Criteria recipientCriteria, Message messageTemplate,
- String controllerMsgId) {
+ MessageId controllerMsgId) {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
@@ -196,9 +198,10 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
String partitionId = taskQueueName + "_" + existingTopPartitionId;
existingTopPartitionId++;
String instanceName = task.getTgtName();
- newAddedScheduledTasks.setPartitionState(partitionId, instanceName, "COMPLETED");
+ newAddedScheduledTasks.setPartitionState(Id.partition(partitionId),
+ Id.participant(instanceName), State.from("COMPLETED"));
task.getRecord().setSimpleField(instanceName, "COMPLETED");
- task.getRecord().setSimpleField(CONTROLLER_MSG_ID, controllerMsgId);
+ task.getRecord().setSimpleField(CONTROLLER_MSG_ID, controllerMsgId.stringify());
List<String> priorityList = new LinkedList<String>();
priorityList.add(instanceName);
@@ -206,7 +209,7 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
newAddedScheduledTasks.getRecord().setMapField(partitionId,
task.getRecord().getSimpleFields());
_logger.info("Scheduling for controllerMsg " + controllerMsgId + " , sending task "
- + partitionId + " " + task.getMsgIdString() + " to " + instanceName);
+ + partitionId + " " + task.getMsgId() + " to " + instanceName);
if (_logger.isDebugEnabled()) {
_logger.debug(task.getRecord().getSimpleFields());
@@ -222,12 +225,12 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
ZNRecord statusUpdate =
accessor.getProperty(
- keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgIdString())).getRecord();
+ keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(), _message
+ .getMsgId().stringify())).getRecord();
statusUpdate.getMapFields().put("SentMessageCount", sendSummary);
accessor.updateProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgIdString()), new StatusUpdate(statusUpdate));
+ _message.getMsgId().stringify()), new StatusUpdate(statusUpdate));
}
private int findTopPartitionId(IdealState currentTaskQueue) {
@@ -251,7 +254,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.getMsgIdString()
+ throw new HelixException("Unexpected msg type for message " + _message.getMsgId()
+ " type:" + _message.getMsgType());
}
// Parse timeout value
@@ -298,11 +301,11 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
if (InstanceType.PARTICIPANT == recipientCriteria.getRecipientInstanceType()
&& hasSchedulerTaskQueue) {
handleMessageUsingScheduledTaskQueue(recipientCriteria, messageTemplate,
- _message.getMsgIdString());
+ _message.getMsgId());
result.setSuccess(true);
- result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgIdString());
+ result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId().stringify());
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
+ "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
return result;
}
@@ -327,24 +330,24 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
ZNRecord statusUpdate =
accessor.getProperty(
- keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgIdString())).getRecord();
+ keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(), _message
+ .getMsgId().stringify())).getRecord();
statusUpdate.getMapFields().put("SentMessageCount", sendSummary);
accessor.setProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
- _message.getMsgIdString()), new StatusUpdate(statusUpdate));
+ _message.getMsgId().stringify()), new StatusUpdate(statusUpdate));
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
- result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgIdString());
+ "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
+ result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId().stringify());
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.getMsgIdString(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 71a5c4a..0a005f2 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
@@ -22,11 +22,11 @@ package org.apache.helix.manager.zk;
import java.util.ArrayList;
import java.util.List;
+import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.HelixException;
import org.apache.helix.HelixTimerTask;
import org.apache.helix.InstanceType;
import org.apache.helix.PreConnectCallback;
-import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.controller.GenericHelixController;
import org.apache.helix.healthcheck.HealthStatsAggregationTask;
import org.apache.helix.healthcheck.HealthStatsAggregator;
@@ -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.getSessionIdString();
+ String sessionId = leader.getSessionId().stringify();
if (leaderName != null && leaderName.equals(_instanceName) && sessionId != null
&& sessionId.equals(_sessionId)) {
return true;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 af33b03..3cf9244 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
@@ -26,8 +26,8 @@ import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
-import org.apache.helix.PropertyType;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.PropertyType;
import org.apache.helix.controller.GenericHelixController;
import org.apache.helix.controller.restlet.ZKPropertyTransferServer;
import org.apache.helix.model.LeaderHistory;
@@ -138,7 +138,7 @@ public class DistributedLeaderElection 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/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 ff932a9..743fcc3 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
@@ -27,7 +27,6 @@ import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.helix.AccessOption;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
import org.apache.helix.InstanceType;
@@ -36,13 +35,12 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.messaging.DefaultMessagingService;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.HelixConfigScope;
+import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
-import org.apache.helix.model.StateModelDefinition;
-import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
-import org.apache.helix.participant.HelixStateMachineEngine;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.participant.statemachine.ScheduledTaskStateModelFactory;
import org.apache.log4j.Logger;
@@ -151,14 +149,14 @@ public class ParticipantManagerHelper {
* update sessionId field in live-instance if necessary
*/
LiveInstance curLiveInstance = new LiveInstance(record);
- if (!curLiveInstance.getSessionIdString().equals(_sessionId)) {
+ if (!curLiveInstance.getSessionId().stringify().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.getSessionIdString() + ", new-sessionId: "
+ + ", old-sessionId: " + curLiveInstance.getSessionId() + ", new-sessionId: "
+ _sessionId);
curLiveInstance.setSessionId(_sessionId);
@@ -230,9 +228,10 @@ public class ParticipantManagerHelper {
String curStatePath =
_keyBuilder.currentState(_instanceName, _sessionId, lastCurState.getResourceName())
.getPath();
- _dataAccessor.getBaseDataAccessor().update(curStatePath,
- new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialStateString(), lastCurState),
- AccessOption.PERSISTENT);
+ _dataAccessor.getBaseDataAccessor().update(
+ curStatePath,
+ new CurStateCarryOverUpdater(_sessionId, stateModel.getInitialStateString(),
+ lastCurState), AccessOption.PERSISTENT);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 7c898ac..89c412c 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
@@ -53,6 +53,11 @@ import org.apache.helix.PropertyType;
import org.apache.helix.ZNRecord;
import org.apache.helix.alerts.AlertsHolder;
import org.apache.helix.alerts.StatsHolder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.SessionId;
+import org.apache.helix.api.State;
+import org.apache.helix.api.StateModelDefId;
import org.apache.helix.controller.strategy.DefaultTwoStateStrategy;
import org.apache.helix.model.Alerts;
import org.apache.helix.model.ClusterConstraints;
@@ -326,9 +331,10 @@ public class ZKHelixAdmin implements HelixAdmin {
}
// check partition is in ERROR state
- String sessionId = liveInstance.getSessionIdString();
+ SessionId sessionId = liveInstance.getSessionId();
CurrentState curState =
- accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, resourceName));
+ accessor.getProperty(keyBuilder.currentState(instanceName, sessionId.stringify(),
+ resourceName));
for (String partitionName : resetPartitionNames) {
if (!curState.getState(partitionName).equals(HelixDefinedState.ERROR.toString())) {
throw new HelixException("Can't reset state for " + resourceName + "/" + partitionNames
@@ -337,8 +343,9 @@ public class ZKHelixAdmin implements HelixAdmin {
}
// check stateModelDef exists and get initial state
- String stateModelDef = idealState.getStateModelDefRef();
- StateModelDefinition stateModel = accessor.getProperty(keyBuilder.stateModelDef(stateModelDef));
+ StateModelDefId stateModelDef = idealState.getStateModelDefId();
+ StateModelDefinition stateModel =
+ accessor.getProperty(keyBuilder.stateModelDef(stateModelDef.stringify()));
if (stateModel == null) {
throw new HelixException("Can't reset state for " + resourceName + "/" + partitionNames
+ " on " + instanceName + ", because " + stateModelDef + " is NOT found");
@@ -348,9 +355,9 @@ 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.getTgtSessionIdString())
- || !resourceName.equals(message.getResourceName())
- || !resetPartitionNames.contains(message.getPartitionName())) {
+ || !sessionId.equals(message.getTgtSessionId())
+ || !resourceName.equals(message.getResourceId().stringify())
+ || !resetPartitionNames.contains(message.getPartitionId().stringify())) {
continue;
}
@@ -371,17 +378,17 @@ public class ZKHelixAdmin implements HelixAdmin {
List<PropertyKey> messageKeys = new ArrayList<PropertyKey>();
for (String partitionName : resetPartitionNames) {
// send ERROR to initialState message
- String msgId = UUID.randomUUID().toString();
+ MessageId msgId = Id.message(UUID.randomUUID().toString());
Message message = new Message(MessageType.STATE_TRANSITION, msgId);
message.setSrcName(adminName);
message.setTgtName(instanceName);
message.setMsgState(MessageState.NEW);
- message.setPartitionName(partitionName);
- message.setResourceName(resourceName);
+ message.setPartitionId(Id.partition(partitionName));
+ message.setResourceId(Id.resource(resourceName));
message.setTgtSessionId(sessionId);
message.setStateModelDef(stateModelDef);
- message.setFromState(HelixDefinedState.ERROR.toString());
- message.setToState(stateModel.getInitialStateString());
+ message.setFromState(State.from(HelixDefinedState.ERROR.toString()));
+ message.setToState(stateModel.getInitialState());
message.setStateModelFactoryName(idealState.getStateModelFactoryName());
resetMessages.add(message);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 0f031dc..f9743a4 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.getMsgIdString());
+ _logger.info("OnReply msg " + message.getMsgId());
if (!isDone()) {
_messageReplied.add(message);
try {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 c22abb8..45a6c9a 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
@@ -28,19 +28,21 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
-import org.apache.helix.model.ConfigScope;
-import org.apache.helix.model.builder.ConfigScopeBuilder;
import org.apache.helix.Criteria;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.InstanceType;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
import org.apache.helix.messaging.handling.AsyncCallbackService;
import org.apache.helix.messaging.handling.HelixTaskExecutor;
import org.apache.helix.messaging.handling.MessageHandlerFactory;
+import org.apache.helix.model.ConfigScope;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.builder.ConfigScopeBuilder;
import org.apache.log4j.Logger;
public class DefaultMessagingService implements ClusterMessagingService {
@@ -157,11 +159,12 @@ public class DefaultMessagingService implements ClusterMessagingService {
List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances());
for (LiveInstance liveInstance : liveInstances) {
- sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionIdString());
+ sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionId()
+ .stringify());
}
}
for (Map<String, String> map : matchedList) {
- String id = UUID.randomUUID().toString();
+ MessageId id = Id.message(UUID.randomUUID().toString());
Message newMessage = new Message(message.getRecord(), id);
String srcInstanceName = _manager.getInstanceName();
String tgtInstanceName = map.get("instanceName");
@@ -172,10 +175,10 @@ public class DefaultMessagingService implements ClusterMessagingService {
}
newMessage.setSrcName(srcInstanceName);
newMessage.setTgtName(tgtInstanceName);
- newMessage.setResourceName(map.get("resourceName"));
- newMessage.setPartitionName(map.get("partitionName"));
+ newMessage.setResourceId(Id.resource(map.get("resourceName")));
+ newMessage.setPartitionId(Id.partition(map.get("partitionName")));
if (recipientCriteria.isSessionSpecific()) {
- newMessage.setTgtSessionId(sessionIdMap.get(tgtInstanceName));
+ newMessage.setTgtSessionId(Id.session(sessionIdMap.get(tgtInstanceName)));
}
messages.add(newMessage);
}
@@ -187,7 +190,7 @@ public class DefaultMessagingService implements ClusterMessagingService {
private List<Message> generateMessagesForController(Message message) {
List<Message> messages = new ArrayList<Message>();
- String id = UUID.randomUUID().toString();
+ MessageId id = Id.message(UUID.randomUUID().toString());
Message newMessage = new Message(message.getRecord(), id);
newMessage.setMsgId(id);
newMessage.setSrcName(_manager.getInstanceName());
@@ -262,7 +265,7 @@ public class DefaultMessagingService implements ClusterMessagingService {
public 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();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 8375420..c218a15 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.getMsgIdString() + " type:" + message.getMsgType()
+ "Unexpected msg type for message " + message.getMsgId() + " 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.getMsgIdString() + " does not have correlation id";
+ String errorMsg = "Message " + message.getMsgId() + " 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.getMsgIdString()
+ + message.getMsgId()
+ " 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.getMsgIdString() + " correlation:" + correlationId);
+ _logger.info("Verified reply message " + message.getMsgId() + " 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.getMsgIdString() + ", correlationid:"
+ _logger.info("invoking reply message " + _message.getMsgId() + ", 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.getMsgIdString(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/messaging/handling/BatchMessageHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/BatchMessageHandler.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/BatchMessageHandler.java
index b506148..1c1951a 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/BatchMessageHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/BatchMessageHandler.java
@@ -35,6 +35,8 @@ import org.apache.helix.HelixManager;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.MapKey;
import org.apache.helix.PropertyKey;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.PartitionId;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.Attributes;
@@ -63,11 +65,11 @@ public class BatchMessageHandler extends MessageHandler {
// create sub-messages
_subMessages = new ArrayList<Message>();
- List<String> partitionKeys = _message.getPartitionNames();
- for (String partitionKey : partitionKeys) {
+ List<PartitionId> partitionKeys = _message.getPartitionIds();
+ for (PartitionId partitionKey : partitionKeys) {
// assign a new message id, put batch-msg-id to parent-id field
- Message subMsg = new Message(_message.getRecord(), UUID.randomUUID().toString());
- subMsg.setPartitionName(partitionKey);
+ Message subMsg = new Message(_message.getRecord(), Id.message(UUID.randomUUID().toString()));
+ subMsg.setPartitionId(partitionKey);
subMsg.setAttribute(Attributes.PARENT_MSG_ID, _message.getId());
subMsg.setBatchMessageMode(false);
@@ -131,7 +133,7 @@ public class BatchMessageHandler extends MessageHandler {
preHandleMessage();
int exeBatchSize = 1; // TODO: getExeBatchSize from msg
- List<String> partitionKeys = _message.getPartitionNames();
+ List<PartitionId> partitionKeys = _message.getPartitionIds();
for (int i = 0; i < partitionKeys.size(); i += exeBatchSize) {
if (i + exeBatchSize <= partitionKeys.size()) {
List<Message> msgs = _subMessages.subList(i, i + exeBatchSize);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/messaging/handling/GroupMessageHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/GroupMessageHandler.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/GroupMessageHandler.java
index 6872907..5d67d5c 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/GroupMessageHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/GroupMessageHandler.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.helix.PropertyKey;
+import org.apache.helix.api.PartitionId;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.Attributes;
@@ -52,7 +53,7 @@ public class GroupMessageHandler {
public GroupMessageInfo(Message message) {
_message = message;
- List<String> partitionNames = message.getPartitionNames();
+ List<PartitionId> partitionNames = message.getPartitionIds();
_countDown = new AtomicInteger(partitionNames.size());
_curStateUpdateList = new ConcurrentLinkedQueue<CurrentStateUpdate>();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 4c634b9..b4de571 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
@@ -40,6 +40,10 @@ import org.apache.helix.ZNRecordBucketizer;
import org.apache.helix.ZNRecordDelta;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.ZNRecordDelta.MergeOperation;
+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.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.Attributes;
@@ -92,18 +96,18 @@ public class HelixStateTransitionHandler extends MessageHandler {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
- String partitionName = _message.getPartitionName();
- String fromState = _message.getFromStateString();
+ PartitionId partitionId = _message.getPartitionId();
+ State fromState = _message.getFromState();
// Verify the fromState and current state of the stateModel
- String state = _currentStateDelta.getState(partitionName);
+ String state = _currentStateDelta.getState(partitionId.stringify());
- if (fromState != null && !fromState.equals("*") && !fromState.equalsIgnoreCase(state)) {
+ if (fromState != null && !fromState.equals("*")
+ && !fromState.toString().equalsIgnoreCase(state)) {
String errorMessage =
"Current state of stateModel does not match the fromState in Message"
+ ", Current State:" + state + ", message expected:" + fromState + ", partition: "
- + partitionName + ", from: " + _message.getMsgSrc() + ", to: "
- + _message.getTgtName();
+ + partitionId + ", from: " + _message.getMsgSrc() + ", to: " + _message.getTgtName();
_statusUpdateUtil.logError(_message, HelixStateTransitionHandler.class, errorMessage,
accessor);
@@ -117,9 +121,9 @@ public class HelixStateTransitionHandler extends MessageHandler {
(HelixTaskResult) _notificationContext.get(MapKey.HELIX_TASK_RESULT.toString());
Exception exception = taskResult.getException();
- String partitionKey = _message.getPartitionName();
- String resource = _message.getResourceName();
- String sessionId = _message.getTgtSessionIdString();
+ PartitionId partitionId = _message.getPartitionId();
+ ResourceId resource = _message.getResourceId();
+ SessionId sessionId = _message.getTgtSessionId();
String instanceName = _manager.getInstanceName();
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
@@ -132,18 +136,18 @@ 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.getTgtSessionIdString().equals(_manager.getSessionId())) {
+ if (!_message.getTgtSessionId().stringify().equals(_manager.getSessionId())) {
logger.warn("Session id has changed. Skip postExecutionMessage. Old session "
- + _message.getExecutionSessionIdString() + " , new session : " + _manager.getSessionId());
+ + _message.getExecutionSessionId() + " , new session : " + _manager.getSessionId());
return;
}
if (taskResult.isSuccess()) {
// String fromState = message.getFromState();
- String toState = _message.getToStateString();
- _currentStateDelta.setState(partitionKey, toState);
+ State toState = _message.getToState();
+ _currentStateDelta.setState(partitionId, toState);
- if (toState.equalsIgnoreCase(HelixDefinedState.DROPPED.toString())) {
+ if (toState.toString().equalsIgnoreCase(HelixDefinedState.DROPPED.toString())) {
// for "OnOfflineToDROPPED" message, we need to remove the resource key record
// from the current state of the instance because the resource key is dropped.
// In the state model it will be stayed as "OFFLINE", which is OK.
@@ -155,18 +159,18 @@ public class HelixStateTransitionHandler extends MessageHandler {
List<ZNRecordDelta> deltaList = new ArrayList<ZNRecordDelta>();
deltaList.add(delta);
_currentStateDelta.setDeltaList(deltaList);
- _stateModelFactory.removeStateModel(partitionKey);
+ _stateModelFactory.removeStateModel(partitionId.stringify());
} else {
// if the partition is not to be dropped, update _stateModel to the TO_STATE
- _stateModel.updateState(toState);
+ _stateModel.updateState(toState.toString());
}
} else {
if (exception instanceof HelixStateMismatchException) {
// if fromState mismatch, set current state on zk to stateModel's current state
logger.warn("Force CurrentState on Zk to be stateModel's CurrentState. partitionKey: "
- + partitionKey + ", currentState: " + _stateModel.getCurrentState() + ", message: "
+ + partitionId + ", currentState: " + _stateModel.getCurrentState() + ", message: "
+ _message);
- _currentStateDelta.setState(partitionKey, _stateModel.getCurrentState());
+ _currentStateDelta.setState(partitionId, State.from(_stateModel.getCurrentState()));
} else {
StateTransitionError error =
new StateTransitionError(ErrorType.INTERNAL, ErrorCode.ERROR, exception);
@@ -178,16 +182,16 @@ 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.getMsgIdString());
+ + _message.getPartitionId() + " MsgId : " + _message.getMsgId());
return;
}
}
_stateModel.rollbackOnError(_message, _notificationContext, error);
- _currentStateDelta.setState(partitionKey, HelixDefinedState.ERROR.toString());
+ _currentStateDelta.setState(partitionId, State.from(HelixDefinedState.ERROR.toString()));
_stateModel.updateState(HelixDefinedState.ERROR.toString());
// if we have errors transit from ERROR state, disable the partition
- if (_message.getFromStateString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
+ if (_message.getFromState().toString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
disablePartition();
}
}
@@ -196,8 +200,8 @@ public class HelixStateTransitionHandler extends MessageHandler {
try {
// Update the ZK current state of the node
PropertyKey key =
- keyBuilder.currentState(instanceName, sessionId, resource,
- bucketizer.getBucketName(partitionKey));
+ keyBuilder.currentState(instanceName, sessionId.stringify(), resource.stringify(),
+ bucketizer.getBucketName(partitionId.stringify()));
if (_message.getAttribute(Attributes.PARENT_MSG_ID) == null) {
// normal message
accessor.updateProperty(key, _currentStateDelta);
@@ -206,7 +210,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
ConcurrentHashMap<String, CurrentStateUpdate> csUpdateMap =
(ConcurrentHashMap<String, CurrentStateUpdate>) _notificationContext
.get(MapKey.CURRENT_STATE_UPDATE.toString());
- csUpdateMap.put(partitionKey, new CurrentStateUpdate(key, _currentStateDelta));
+ csUpdateMap.put(partitionId.stringify(), new CurrentStateUpdate(key, _currentStateDelta));
}
} catch (Exception e) {
logger.error("Error when updating current-state ", e);
@@ -220,14 +224,14 @@ public class HelixStateTransitionHandler extends MessageHandler {
void disablePartition() {
String instanceName = _manager.getInstanceName();
- String resourceName = _message.getResourceName();
- String partitionName = _message.getPartitionName();
+ ResourceId resourceId = _message.getResourceId();
+ PartitionId partitionId = _message.getPartitionId();
String clusterName = _manager.getClusterName();
HelixAdmin admin = _manager.getClusterManagmentTool();
- admin.enablePartition(false, clusterName, instanceName, resourceName,
- Arrays.asList(partitionName));
- logger.info("error in transit from ERROR to " + _message.getToStateString() + " for partition: "
- + partitionName + ". disable it on " + instanceName);
+ admin.enablePartition(false, clusterName, instanceName, resourceId.stringify(),
+ Arrays.asList(partitionId.stringify()));
+ logger.info("error in transit from ERROR to " + _message.getToState() + " for partition: "
+ + partitionId + ". disable it on " + instanceName);
}
@@ -256,7 +260,7 @@ public class HelixStateTransitionHandler extends MessageHandler {
taskResult.setException(e);
} catch (Exception e) {
String errorMessage =
- "Exception while executing a state transition task " + message.getPartitionName();
+ "Exception while executing a state transition task " + message.getPartitionId();
logger.error(errorMessage, e);
if (e.getCause() != null && e.getCause() instanceof InterruptedException) {
e = (InterruptedException) e.getCause();
@@ -285,11 +289,11 @@ public class HelixStateTransitionHandler extends MessageHandler {
// by default, we invoke state transition function in state model
Method methodToInvoke = null;
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
methodToInvoke =
- _transitionMethodFinder.getMethodForTransition(_stateModel.getClass(), fromState, toState,
- new Class[] {
+ _transitionMethodFinder.getMethodForTransition(_stateModel.getClass(),
+ fromState.toString(), toState.toString(), new Class[] {
Message.class, NotificationContext.class
});
if (methodToInvoke != null) {
@@ -314,8 +318,8 @@ public class HelixStateTransitionHandler extends MessageHandler {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
String instanceName = _manager.getInstanceName();
- String resourceName = _message.getResourceName();
- String partition = _message.getPartitionName();
+ ResourceId resourceId = _message.getResourceId();
+ PartitionId partition = _message.getPartitionId();
// All internal error has been processed already, so we can skip them
if (type == ErrorType.INTERNAL) {
@@ -327,17 +331,16 @@ public class HelixStateTransitionHandler extends MessageHandler {
// set current state to ERROR for the partition
// if the transition is not canceled, it should go into error state
if (code == ErrorCode.ERROR) {
- CurrentState currentStateDelta = new CurrentState(resourceName);
- currentStateDelta.setState(partition, HelixDefinedState.ERROR.toString());
+ CurrentState currentStateDelta = new CurrentState(resourceId.stringify());
+ currentStateDelta.setState(partition, State.from(HelixDefinedState.ERROR.toString()));
_stateModel.updateState(HelixDefinedState.ERROR.toString());
// if transit from ERROR state, disable the partition
- if (_message.getFromStateString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
+ if (_message.getFromState().toString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
disablePartition();
}
- accessor.updateProperty(
- keyBuilder.currentState(instanceName, _message.getTgtSessionIdString(), resourceName),
- currentStateDelta);
+ accessor.updateProperty(keyBuilder.currentState(instanceName, _message.getTgtSessionId()
+ .stringify(), resourceId.stringify()), currentStateDelta);
}
} finally {
StateTransitionError error = new StateTransitionError(type, code, e);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 dd8612e..39da1aa 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
@@ -19,13 +19,7 @@ package org.apache.helix.messaging.handling;
* under the License.
*/
-import java.util.ArrayList;
import java.util.Date;
-import java.util.List;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.helix.HelixDataAccessor;
@@ -33,12 +27,10 @@ import org.apache.helix.HelixManager;
import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.MapKey;
-import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyKey.Builder;
-import org.apache.helix.messaging.handling.GroupMessageHandler.GroupMessageInfo;
+import org.apache.helix.api.State;
import org.apache.helix.messaging.handling.MessageHandler.ErrorCode;
import org.apache.helix.messaging.handling.MessageHandler.ErrorType;
-import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.Attributes;
import org.apache.helix.model.Message.MessageType;
@@ -98,14 +90,14 @@ public class HelixTask implements MessageTask {
_statusUpdateUtil.logError(_message, HelixTask.class, e,
"State transition interrupted, timeout:" + _isTimeout, accessor);
- logger.info("Message " + _message.getMsgIdString() + " is interrupted");
+ logger.info("Message " + _message.getMsgId() + " 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.getMsgIdString()
+ "Exception while executing a message. " + e + " msgId: " + _message.getMsgId()
+ " type: " + _message.getMsgType();
logger.error(errorMessage, e);
_statusUpdateUtil.logError(_message, HelixTask.class, e, errorMessage, accessor);
@@ -119,17 +111,17 @@ public class HelixTask implements MessageTask {
if (taskResult.isSuccess()) {
_statusUpdateUtil.logInfo(_message, _handler.getClass(),
"Message handling task completed successfully", accessor);
- logger.info("Message " + _message.getMsgIdString() + " completed.");
+ logger.info("Message " + _message.getMsgId() + " completed.");
} else {
type = ErrorType.INTERNAL;
if (taskResult.isInterrupted()) {
- logger.info("Message " + _message.getMsgIdString() + " is interrupted");
+ logger.info("Message " + _message.getMsgId() + " is interrupted");
code = _isTimeout ? ErrorCode.TIMEOUT : ErrorCode.CANCEL;
if (_isTimeout) {
int retryCount = _message.getRetryCount();
logger.info("Message timeout, retry count: " + retryCount + " msgId:"
- + _message.getMsgIdString());
+ + _message.getMsgId());
_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 +158,12 @@ public class HelixTask implements MessageTask {
code = ErrorCode.ERROR;
String errorMessage =
- "Exception after executing a message, msgId: " + _message.getMsgIdString() + e;
+ "Exception after executing a message, msgId: " + _message.getMsgId() + e;
logger.error(errorMessage, e);
_statusUpdateUtil.logError(_message, HelixTask.class, errorMessage, accessor);
} finally {
long end = System.currentTimeMillis();
- logger.info("msg: " + _message.getMsgIdString() + " handling task completed, results:"
+ logger.info("msg: " + _message.getMsgId() + " 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 +182,10 @@ public class HelixTask implements MessageTask {
Builder keyBuilder = accessor.keyBuilder();
if (message.getTgtName().equalsIgnoreCase("controller")) {
// TODO: removeProperty returns boolean
- accessor.removeProperty(keyBuilder.controllerMessage(message.getMsgIdString()));
+ accessor.removeProperty(keyBuilder.controllerMessage(message.getMsgId().stringify()));
} else {
- accessor.removeProperty(keyBuilder.message(_manager.getInstanceName(), message.getMsgIdString()));
+ accessor.removeProperty(keyBuilder.message(_manager.getInstanceName(), message.getMsgId()
+ .stringify()));
}
}
@@ -214,11 +207,13 @@ public class HelixTask implements MessageTask {
if (message.getSrcInstanceType() == InstanceType.PARTICIPANT) {
Builder keyBuilder = accessor.keyBuilder();
- accessor.setProperty(keyBuilder.message(message.getMsgSrc(), replyMessage.getMsgIdString()),
+ accessor.setProperty(
+ keyBuilder.message(message.getMsgSrc(), replyMessage.getMsgId().stringify()),
replyMessage);
} else if (message.getSrcInstanceType() == InstanceType.CONTROLLER) {
Builder keyBuilder = accessor.keyBuilder();
- accessor.setProperty(keyBuilder.controllerMessage(replyMessage.getMsgIdString()), replyMessage);
+ accessor.setProperty(keyBuilder.controllerMessage(replyMessage.getMsgId().stringify()),
+ replyMessage);
}
_statusUpdateUtil.logInfo(message, HelixTask.class,
"1 msg replied to " + replyMessage.getTgtName(), accessor);
@@ -237,13 +232,13 @@ public class HelixTask implements MessageTask {
long totalDelay = now - msgReadTime;
long executionDelay = now - msgExecutionStartTime;
if (totalDelay > 0 && executionDelay > 0) {
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
String transition = fromState + "--" + toState;
StateTransitionContext cxt =
- new StateTransitionContext(manager.getClusterName(), manager.getInstanceName(),
- message.getResourceName(), transition);
+ new StateTransitionContext(manager.getClusterName(), manager.getInstanceName(), message
+ .getResourceId().stringify(), transition);
StateTransitionDataPoint data =
new StateTransitionDataPoint(totalDelay, executionDelay, taskResult.isSuccess());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 276c27f..f9f96ac 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
@@ -35,8 +35,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.helix.ConfigAccessor;
-import org.apache.helix.model.ConfigScope;
-import org.apache.helix.model.builder.ConfigScopeBuilder;
import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
@@ -44,13 +42,18 @@ import org.apache.helix.HelixManager;
import org.apache.helix.MessageListener;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.MapKey;
-import org.apache.helix.PropertyKey;
import org.apache.helix.NotificationContext.Type;
+import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ResourceId;
+import org.apache.helix.api.SessionId;
+import org.apache.helix.model.ConfigScope;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageState;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.builder.ConfigScopeBuilder;
import org.apache.helix.monitoring.ParticipantMonitor;
import org.apache.helix.participant.HelixStateMachineEngine;
import org.apache.helix.util.StatusUpdateUtil;
@@ -161,9 +164,9 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
ExecutorService findExecutorServiceForMsg(Message message) {
ExecutorService executorService = _executorMap.get(message.getMsgType());
if (message.getMsgType().equals(MessageType.STATE_TRANSITION.toString())) {
- String resourceName = message.getResourceName();
- if (resourceName != null) {
- String key = message.getMsgType() + "." + resourceName;
+ ResourceId resourceId = message.getResourceId();
+ if (resourceId != null) {
+ String key = message.getMsgType() + "." + resourceId;
if (_executorMap.containsKey(key)) {
LOG.info("Find per-resource thread pool with key: " + key);
executorService = _executorMap.get(key);
@@ -223,7 +226,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
try {
if (message.getMsgType().equals(MessageType.STATE_TRANSITION.toString())) {
- checkResourceConfig(message.getResourceName(), notificationContext.getManager());
+ checkResourceConfig(message.getResourceId().toString(), notificationContext.getManager());
}
LOG.info("Scheduling message: " + taskId);
@@ -430,14 +433,14 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
continue;
}
- String tgtSessionId = message.getTgtSessionIdString();
+ SessionId tgtSessionId = message.getTgtSessionId();
// sessionId mismatch normally means message comes from expired session, just remove it
- if (!sessionId.equals(tgtSessionId) && !tgtSessionId.equals("*")) {
+ if (!sessionId.equals(tgtSessionId.toString()) && !tgtSessionId.toString().equals("*")) {
String warningMessage =
"SessionId does NOT match. expected sessionId: " + sessionId
+ ", tgtSessionId in message: " + tgtSessionId + ", messageId: "
- + message.getMsgIdString();
+ + message.getMsgId();
LOG.warn(warningMessage);
accessor.removeProperty(message.getKey(keyBuilder, instanceName));
_statusUpdateUtil.logWarning(message, HelixStateMachineEngine.class, warningMessage,
@@ -452,7 +455,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.getMsgIdString());
+ LOG.trace("Message already read. msgId: " + message.getMsgId());
}
continue;
}
@@ -465,9 +468,9 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
}
handlers.add(createHandler);
} catch (Exception e) {
- LOG.error("Failed to create message handler for " + message.getMsgIdString(), e);
+ LOG.error("Failed to create message handler for " + message.getMsgId(), e);
String error =
- "Failed to create message handler for " + message.getMsgIdString() + ", exception: " + e;
+ "Failed to create message handler for " + message.getMsgId() + ", exception: " + e;
_statusUpdateUtil.logError(message, HelixStateMachineEngine.class, e, error, accessor);
@@ -481,7 +484,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
// update msgState to read
message.setMsgState(MessageState.READ);
message.setReadTimeStamp(new Date().getTime());
- message.setExecuteSessionId(changeContext.getManager().getSessionId());
+ message.setExecuteSessionId(Id.session(changeContext.getManager().getSessionId()));
_statusUpdateUtil.logInfo(message, HelixStateMachineEngine.class, "New Message", accessor);
@@ -491,15 +494,17 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
// do it for non-controller and state transition messages only
if (!message.isControlerMsg()
&& message.getMsgType().equals(Message.MessageType.STATE_TRANSITION.toString())) {
- String resourceName = message.getResourceName();
- if (!curResourceNames.contains(resourceName) && !createCurStateNames.contains(resourceName)) {
- createCurStateNames.add(resourceName);
- createCurStateKeys.add(keyBuilder.currentState(instanceName, sessionId, resourceName));
-
- CurrentState metaCurState = new CurrentState(resourceName);
+ ResourceId resourceId = message.getResourceId();
+ if (!curResourceNames.contains(resourceId.stringify())
+ && !createCurStateNames.contains(resourceId.stringify())) {
+ createCurStateNames.add(resourceId.stringify());
+ createCurStateKeys.add(keyBuilder.currentState(instanceName, sessionId,
+ resourceId.stringify()));
+
+ CurrentState metaCurState = new CurrentState(resourceId.stringify());
metaCurState.setBucketSize(message.getBucketSize());
metaCurState.setStateModelDefRef(message.getStateModelDef());
- metaCurState.setSessionId(sessionId);
+ metaCurState.setSessionId(Id.session(sessionId));
metaCurState.setBatchMessageMode(message.getBatchMessageMode());
String ftyName = message.getStateModelFactoryName();
if (ftyName != null) {
@@ -543,7 +548,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.getMsgIdString());
+ + message.getMsgId());
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 17f525e..e1b4f0f 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.getMsgIdString() + " timeout : "
+ LOG.warn("Message time out, canceling. id:" + message.getMsgId() + " timeout : "
+ message.getExecutionTimeout());
_task.onTimeout();
_executor.cancelTask(_task);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 3b7557d..3b46c13 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,11 +164,12 @@ public class ClusterConstraints extends HelixProperty {
String msgType = msg.getMsgType();
attributes.put(ConstraintAttribute.MESSAGE_TYPE, msgType);
if (MessageType.STATE_TRANSITION.toString().equals(msgType)) {
- if (msg.getFromStateString() != null && msg.getToStateString() != null) {
- attributes.put(ConstraintAttribute.TRANSITION, msg.getFromStateString() + "-" + msg.getToStateString());
+ if (msg.getFromState() != null && msg.getToState() != null) {
+ attributes.put(ConstraintAttribute.TRANSITION,
+ Transition.from(msg.getFromState(), msg.getToState()).toString());
}
- if (msg.getResourceName() != null) {
- attributes.put(ConstraintAttribute.RESOURCE, msg.getResourceName());
+ if (msg.getResourceId() != null) {
+ attributes.put(ConstraintAttribute.RESOURCE, msg.getResourceId().stringify());
}
if (msg.getTgtName() != null) {
attributes.put(ConstraintAttribute.INSTANCE, msg.getTgtName());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 dba439b..0797145 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
@@ -117,26 +117,18 @@ public class CurrentState extends HelixProperty {
/**
* Get the session that this current state corresponds to
- * @return String session identifier
- */
- 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());
+ return Id.session(_record.getSimpleField(CurrentStateProperty.SESSION_ID.toString()));
}
/**
* Set the session that this current state corresponds to
- * @param sessionId String session identifier
+ * @param sessionId session identifier
*/
- public void setSessionId(String sessionId) {
- _record.setSimpleField(CurrentStateProperty.SESSION_ID.toString(), sessionId);
+ public void setSessionId(SessionId sessionId) {
+ _record.setSimpleField(CurrentStateProperty.SESSION_ID.toString(), sessionId.stringify());
}
/**
@@ -179,6 +171,15 @@ public class CurrentState extends HelixProperty {
}
/**
+ * Set the state model that the resource follows
+ * @param stateModelName an identifier of the state model
+ */
+ public void setStateModelDefId(StateModelDefId stateModelId) {
+ _record.setSimpleField(CurrentStateProperty.STATE_MODEL_DEF.toString(),
+ stateModelId.stringify());
+ }
+
+ /**
* Get the state model that the resource follows
* @return an identifier of the state model
*/
@@ -191,12 +192,13 @@ public class CurrentState extends HelixProperty {
* @param partitionName the name of the partition
* @param state the state of the partition
*/
- public void setState(String partitionName, String state) {
+ public void setState(PartitionId partitionId, State state) {
Map<String, Map<String, String>> mapFields = _record.getMapFields();
- if (mapFields.get(partitionName) == null) {
- mapFields.put(partitionName, new TreeMap<String, String>());
+ if (mapFields.get(partitionId.stringify()) == null) {
+ mapFields.put(partitionId.stringify(), new TreeMap<String, String>());
}
- mapFields.get(partitionName).put(CurrentStateProperty.CURRENT_STATE.toString(), state);
+ mapFields.get(partitionId.stringify()).put(CurrentStateProperty.CURRENT_STATE.toString(),
+ state.toString());
}
/**
@@ -250,11 +252,37 @@ public class CurrentState extends HelixProperty {
LOG.error("Current state does not contain state model ref. id:" + getResourceName());
return false;
}
- if (getSessionIdString() == null) {
+ if (getSessionId() == null) {
LOG.error("CurrentState does not contain session id, id : " + getResourceName());
return false;
}
return true;
}
+ /**
+ * Convert a string map to a concrete partition map
+ * @param rawMap map of partition name to state name
+ * @return map of partition id to state
+ */
+ public static Map<PartitionId, State> partitionStateMapFromStringMap(Map<String, String> rawMap) {
+ Map<PartitionId, State> partitionStateMap = new HashMap<PartitionId, State>();
+ for (String partitionId : rawMap.keySet()) {
+ partitionStateMap.put(Id.partition(partitionId), State.from(rawMap.get(partitionId)));
+ }
+ return partitionStateMap;
+ }
+
+ /**
+ * Convert a partition map to a string map
+ * @param partitionStateMap map of partition id to state
+ * @return map of partition name to state name
+ */
+ public static Map<String, String> stringMapFromPartitionStateMap(
+ Map<PartitionId, State> partitionStateMap) {
+ Map<String, String> rawMap = new HashMap<String, String>();
+ for (PartitionId partitionId : partitionStateMap.keySet()) {
+ rawMap.put(partitionId.stringify(), partitionStateMap.get(partitionId).toString());
+ }
+ return rawMap;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 c7ff8bb..ebbce23 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
@@ -70,6 +70,20 @@ public class ExternalView extends HelixProperty {
}
/**
+ * For a given replica, specify which partition it corresponds to, where it is served, and its
+ * current state
+ * @param partitionId the partition of the replica being served
+ * @param participantId the instance serving the replica
+ * @param state the state the replica is in
+ */
+ public void setState(PartitionId partitionId, ParticipantId participantId, State state) {
+ if (_record.getMapField(partitionId.stringify()) == null) {
+ _record.setMapField(partitionId.stringify(), new TreeMap<String, String>());
+ }
+ _record.getMapField(partitionId.stringify()).put(participantId.stringify(), state.toString());
+ }
+
+ /**
* For a given partition, indicate where and in what state each of its replicas is in
* @param partitionName the partition to set
* @param currentStateMap (instance, state) pairs
@@ -79,6 +93,17 @@ public class ExternalView extends HelixProperty {
}
/**
+ * For a given partition, indicate where and in what state each of its replicas is in
+ * @param partitionId the partition to set
+ * @param currentStateMap (participant, state) pairs
+ */
+ public void setStateMap(PartitionId partitionId, Map<ParticipantId, State> currentStateMap) {
+ for (ParticipantId participantId : currentStateMap.keySet()) {
+ setState(partitionId, participantId, currentStateMap.get(participantId));
+ }
+ }
+
+ /**
* Get all the partitions of the resource
* @return a set of partition names
*/
@@ -142,4 +167,13 @@ public class ExternalView extends HelixProperty {
public boolean isValid() {
return true;
}
+
+ /**
+ * Convert a partition mapping as strings into a participant state map
+ * @param rawMap the map of participant name to state
+ * @return converted map
+ */
+ public static Map<ParticipantId, State> stateMapFromStringMap(Map<String, String> rawMap) {
+ return ResourceAssignment.replicaMapFromStringMap(rawMap);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 cbf7a64..fcadaf3 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
@@ -21,6 +21,7 @@ package org.apache.helix.model;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -40,9 +41,11 @@ import org.apache.helix.api.StateModelDefId;
import org.apache.helix.controller.rebalancer.Rebalancer;
import org.apache.log4j.Logger;
+import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
/**
* The ideal states of all partitions in a resource
@@ -173,8 +176,16 @@ public class IdealState extends HelixProperty {
}
/**
+ * Set a reference to the user-defined rebalancer associated with this resource(if any)
+ * @param rebalancerRef a reference to a user-defined rebalancer
+ */
+ public void setRebalancerRef(RebalancerRef rebalancerRef) {
+ setRebalancerClassName(rebalancerRef.toString());
+ }
+
+ /**
* Get a reference to the user-defined rebalancer associated with this resource(if any)
- * @return
+ * @return RebalancerRef
*/
public RebalancerRef getRebalancerRef() {
String className = getRebalancerClassName();
@@ -232,6 +243,20 @@ public class IdealState extends HelixProperty {
}
/**
+ * Set the preferred participant placement and state for a partition replica
+ * @param partitionId the replica to set
+ * @param participantId the assigned participant
+ * @param state the replica state in this instance
+ */
+ public void setPartitionState(PartitionId partitionId, ParticipantId participantId, State state) {
+ Map<String, String> mapField = _record.getMapField(partitionId.stringify());
+ if (mapField == null) {
+ _record.setMapField(partitionId.stringify(), new TreeMap<String, String>());
+ }
+ _record.getMapField(partitionId.stringify()).put(participantId.stringify(), state.toString());
+ }
+
+ /**
* Get all of the partitions
* @return a set of partition names
*/
@@ -279,6 +304,20 @@ public class IdealState extends HelixProperty {
}
/**
+ * Set the current mapping of a partition
+ * @param partitionId the partition to set
+ * @param participantStateMap (participant id, state) pairs
+ */
+ public void setParticipantStateMap(PartitionId partitionId,
+ Map<ParticipantId, State> participantStateMap) {
+ Map<String, String> rawMap = new HashMap<String, String>();
+ for (ParticipantId participantId : participantStateMap.keySet()) {
+ rawMap.put(participantId.stringify(), participantStateMap.get(participantId).toString());
+ }
+ _record.setMapField(partitionId.stringify(), rawMap);
+ }
+
+ /**
* 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)
@@ -346,6 +385,19 @@ public class IdealState extends HelixProperty {
}
/**
+ * Set the preference list of a partition
+ * @param partitionId the id of the partition to set
+ * @param preferenceList a list of participants that can serve replicas of the partition
+ */
+ public void setPreferenceList(PartitionId partitionId, List<ParticipantId> preferenceList) {
+ List<String> rawPreferenceList = new ArrayList<String>();
+ for (ParticipantId participantId : preferenceList) {
+ rawPreferenceList.add(participantId.stringify());
+ }
+ _record.setListField(partitionId.stringify(), rawPreferenceList);
+ }
+
+ /**
* Get the preference list of a partition
* @param partitionName the name of the partition
* @return a list of instances that can serve replicas of the partition
@@ -556,10 +608,10 @@ public class IdealState extends HelixProperty {
public void updateFromAssignment(ResourceAssignment assignment) {
_record.getMapFields().clear();
_record.getListFields().clear();
- for (Partition partition : assignment.getMappedPartitions()) {
- Map<String, String> replicaMap = assignment.getReplicaMap(partition);
- setInstanceStateMap(partition.getPartitionName(), replicaMap);
- setPreferenceList(partition.getPartitionName(), new ArrayList<String>(replicaMap.keySet()));
+ for (PartitionId partition : assignment.getMappedPartitions()) {
+ Map<ParticipantId, State> replicaMap = assignment.getReplicaMap(partition);
+ setParticipantStateMap(partition, replicaMap);
+ setPreferenceList(partition, new ArrayList<ParticipantId>(replicaMap.keySet()));
}
}
@@ -604,4 +656,114 @@ public class IdealState extends HelixProperty {
}
return property;
}
+
+ /**
+ * Convert a preference list of strings into a preference list of participants
+ * @param rawPreferenceList the list of strings representing participant names
+ * @return converted list
+ */
+ public static List<ParticipantId> preferenceListFromStringList(List<String> rawPreferenceList) {
+ return Lists.transform(rawPreferenceList, new Function<String, ParticipantId>() {
+ @Override
+ public ParticipantId apply(String participantName) {
+ return Id.participant(participantName);
+ }
+ });
+ }
+
+ /**
+ * Convert preference lists of strings into preference lists of participants
+ * @param rawPreferenceLists a map of partition name to a list of participant names
+ * @return converted lists as a map
+ */
+ public static Map<PartitionId, List<ParticipantId>> preferenceListsFromStringLists(
+ Map<String, List<String>> rawPreferenceLists) {
+ Map<PartitionId, List<ParticipantId>> preferenceLists =
+ new HashMap<PartitionId, List<ParticipantId>>();
+ for (String partitionId : rawPreferenceLists.keySet()) {
+ preferenceLists.put(Id.partition(partitionId),
+ preferenceListFromStringList(rawPreferenceLists.get(partitionId)));
+ }
+ return preferenceLists;
+ }
+
+ /**
+ * Convert a preference list of participants into a preference list of strings
+ * @param preferenceList the list of strings representing participant ids
+ * @return converted list
+ */
+ public static List<String> stringListFromPreferenceList(List<ParticipantId> preferenceList) {
+ return Lists.transform(preferenceList, new Function<ParticipantId, String>() {
+ @Override
+ public String apply(ParticipantId participantId) {
+ return participantId.stringify();
+ }
+ });
+ }
+
+ /**
+ * Convert preference lists of participants into preference lists of strings
+ * @param preferenceLists a map of partition id to a list of participant ids
+ * @return converted lists as a map
+ */
+ public static Map<String, List<String>> stringListsFromPreferenceLists(
+ Map<PartitionId, List<ParticipantId>> preferenceLists) {
+ Map<String, List<String>> rawPreferenceLists = new HashMap<String, List<String>>();
+ for (PartitionId partitionId : preferenceLists.keySet()) {
+ rawPreferenceLists.put(partitionId.stringify(),
+ stringListFromPreferenceList(preferenceLists.get(partitionId)));
+ }
+ return rawPreferenceLists;
+ }
+
+ /**
+ * Convert a partition mapping as strings into a participant state map
+ * @param rawMap the map of participant name to state
+ * @return converted map
+ */
+ public static Map<ParticipantId, State> participantStateMapFromStringMap(
+ Map<String, String> rawMap) {
+ return ResourceAssignment.replicaMapFromStringMap(rawMap);
+ }
+
+ /**
+ * Convert a full state mapping as strings into participant state maps
+ * @param rawMap the map of partition name to participant name and state
+ * @return converted maps
+ */
+ public static Map<PartitionId, Map<ParticipantId, State>> participantStateMapsFromStringMaps(
+ Map<String, Map<String, String>> rawMaps) {
+ Map<PartitionId, Map<ParticipantId, State>> participantStateMaps =
+ new HashMap<PartitionId, Map<ParticipantId, State>>();
+ for (String partitionId : rawMaps.keySet()) {
+ participantStateMaps.put(Id.partition(partitionId),
+ participantStateMapFromStringMap(rawMaps.get(partitionId)));
+ }
+ return participantStateMaps;
+ }
+
+ /**
+ * Convert a partition mapping into a mapping of string names
+ * @param participantStateMap the map of participant id to state
+ * @return converted map
+ */
+ public static Map<String, String> stringMapFromParticipantStateMap(
+ Map<ParticipantId, State> participantStateMap) {
+ return ResourceAssignment.stringMapFromReplicaMap(participantStateMap);
+ }
+
+ /**
+ * Convert a full state mapping into a mapping of string names
+ * @param participantStateMaps the map of partition id to participant id and state
+ * @return converted maps
+ */
+ public static Map<String, Map<String, String>> stringMapsFromParticipantStateMaps(
+ Map<PartitionId, Map<ParticipantId, State>> participantStateMaps) {
+ Map<String, Map<String, String>> rawMaps = new HashMap<String, Map<String, String>>();
+ for (PartitionId partitionId : participantStateMaps.keySet()) {
+ rawMaps.put(partitionId.stringify(),
+ stringMapFromParticipantStateMap(participantStateMaps.get(partitionId)));
+ }
+ return rawMaps;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 2905965..032e5c2 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
@@ -29,6 +29,7 @@ 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.log4j.Logger;
/**
@@ -215,6 +216,15 @@ public class InstanceConfig extends HelixProperty {
_record.setListField(InstanceConfigProperty.HELIX_DISABLED_PARTITION.toString(), list);
}
+ /**
+ * Set the enabled state for a partition on this instance
+ * @param partitionId the partition to set
+ * @param enabled true to enable, false to disable
+ */
+ public void setInstanceEnabledForPartition(PartitionId partitionId, boolean enabled) {
+ setInstanceEnabledForPartition(partitionId.stringify(), enabled);
+ }
+
@Override
public boolean equals(Object obj) {
if (obj instanceof InstanceConfig) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/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 86eb2d8..4e0724a 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
@@ -69,19 +69,11 @@ public class LiveInstance extends HelixProperty {
}
/**
- * Get the session that this instance corresponds to
- * @return session identifier
- */
- 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());
+ return Id.session(_record.getSimpleField(LiveInstanceProperty.SESSION_ID.toString()));
}
/**
@@ -101,19 +93,11 @@ public class LiveInstance extends HelixProperty {
}
/**
- * Get the version of Helix that this instance is running
- * @return the version
- */
- 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());
+ return HelixVersion.from(_record.getSimpleField(LiveInstanceProperty.HELIX_VERSION.toString()));
}
/**
@@ -174,11 +158,11 @@ public class LiveInstance extends HelixProperty {
@Override
public boolean isValid() {
- if (getSessionIdString() == null) {
+ if (getSessionId() == null) {
_logger.error("liveInstance does not have session id. id:" + _record.getId());
return false;
}
- if (getHelixVersionString() == null) {
+ if (getHelixVersion() == null) {
_logger.error("liveInstance does not have CLM verion. id:" + _record.getId());
return false;
}