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:31 UTC
[5/5] git commit: [HELIX-109] Review Helix model package, rb=13878
[HELIX-109] Review Helix model package, rb=13878
Project: http://git-wip-us.apache.org/repos/asf/incubator-helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-helix/commit/3be06fa6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-helix/tree/3be06fa6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-helix/diff/3be06fa6
Branch: refs/heads/helix-logical-model
Commit: 3be06fa66cfa418f4ebc41f879c9313a91d60dbf
Parents: becfda1
Author: zzhang <zz...@uci.edu>
Authored: Fri Aug 30 10:16:17 2013 -0700
Committer: zzhang <zz...@uci.edu>
Committed: Fri Aug 30 10:16:17 2013 -0700
----------------------------------------------------------------------
.../webapp/resources/IdealStateResource.java | 6 +-
.../webapp/resources/ResourceGroupResource.java | 2 +-
.../resources/SchedulerTasksResource.java | 15 +-
.../helix/tools/TestResetPartitionState.java | 11 +-
.../org/apache/helix/agent/AgentStateModel.java | 24 ++-
.../java/org/apache/helix/api/Controller.java | 4 +-
.../java/org/apache/helix/api/CurState.java | 78 --------
.../main/java/org/apache/helix/api/ExtView.java | 62 ------
.../java/org/apache/helix/api/HelixVersion.java | 3 +
.../src/main/java/org/apache/helix/api/Id.java | 14 +-
.../src/main/java/org/apache/helix/api/Msg.java | 68 -------
.../apache/helix/api/ParticipantAccessor.java | 10 +-
.../java/org/apache/helix/api/PartitionId.java | 10 +
.../org/apache/helix/api/RebalancerConfig.java | 11 +-
.../org/apache/helix/api/RebalancerRef.java | 2 +
.../java/org/apache/helix/api/Resource.java | 11 +-
.../org/apache/helix/api/ResourceAccessor.java | 4 +-
.../org/apache/helix/api/RscAssignment.java | 140 --------------
.../main/java/org/apache/helix/api/State.java | 7 +
.../controller/GenericHelixController.java | 2 +-
.../controller/rebalancer/AutoRebalancer.java | 7 +-
.../controller/rebalancer/CustomRebalancer.java | 7 +-
.../rebalancer/SemiAutoRebalancer.java | 7 +-
.../stages/BestPossibleStateCalcStage.java | 8 +-
.../stages/BestPossibleStateOutput.java | 12 ++
.../controller/stages/ClusterDataCache.java | 8 +-
.../stages/CompatibilityCheckStage.java | 4 +-
.../stages/CurrentStateComputationStage.java | 32 ++--
.../stages/MessageGenerationPhase.java | 25 +--
.../stages/MessageSelectionStage.java | 11 +-
.../stages/ResourceComputationStage.java | 6 +-
.../controller/stages/TaskAssignmentStage.java | 21 ++-
.../helix/manager/zk/ControllerManager.java | 2 +-
.../manager/zk/CurStateCarryOverUpdater.java | 11 +-
.../DefaultControllerMessageHandlerFactory.java | 8 +-
...ltParticipantErrorMessageHandlerFactory.java | 12 +-
.../DefaultSchedulerMessageHandlerFactory.java | 55 +++---
.../zk/DistributedControllerManager.java | 4 +-
.../manager/zk/DistributedLeaderElection.java | 4 +-
.../manager/zk/ParticipantManagerHelper.java | 17 +-
.../apache/helix/manager/zk/ZKHelixAdmin.java | 31 +--
.../apache/helix/messaging/AsyncCallback.java | 2 +-
.../messaging/DefaultMessagingService.java | 21 ++-
.../handling/AsyncCallbackService.java | 12 +-
.../messaging/handling/BatchMessageHandler.java | 12 +-
.../messaging/handling/GroupMessageHandler.java | 3 +-
.../handling/HelixStateTransitionHandler.java | 89 ++++-----
.../helix/messaging/handling/HelixTask.java | 43 ++---
.../messaging/handling/HelixTaskExecutor.java | 49 ++---
.../messaging/handling/MessageTimeoutTask.java | 2 +-
.../apache/helix/model/ClusterConstraints.java | 9 +-
.../org/apache/helix/model/CurrentState.java | 62 ++++--
.../org/apache/helix/model/ExternalView.java | 34 ++++
.../java/org/apache/helix/model/IdealState.java | 172 ++++++++++++++++-
.../org/apache/helix/model/InstanceConfig.java | 10 +
.../org/apache/helix/model/LiveInstance.java | 24 +--
.../java/org/apache/helix/model/Message.java | 189 ++++++-------------
.../apache/helix/model/ResourceAssignment.java | 87 ++++++---
.../helix/model/StateModelDefinition.java | 2 +-
.../java/org/apache/helix/model/Transition.java | 22 ++-
.../builder/StateTransitionTableBuilder.java | 19 +-
.../DistClusterControllerElection.java | 2 +-
.../DistClusterControllerStateModel.java | 8 +-
.../participant/HelixStateMachineEngine.java | 57 +++---
.../org/apache/helix/tools/MessagePoster.java | 21 ++-
.../helix/tools/StateModelConfigGenerator.java | 8 +-
.../org/apache/helix/tools/ZkLogAnalyzer.java | 9 +-
.../org/apache/helix/util/StatusUpdateUtil.java | 46 ++---
.../org/apache/helix/TestHelixTaskExecutor.java | 19 +-
.../org/apache/helix/TestHelixTaskHandler.java | 43 +++--
.../test/java/org/apache/helix/TestHelper.java | 15 +-
.../java/org/apache/helix/TestZKCallback.java | 19 +-
.../java/org/apache/helix/ZkUnitTestBase.java | 10 +-
.../TestCurrentStateComputationStage.java | 28 ++-
.../stages/TestMessageThrottleStage.java | 44 ++---
.../stages/TestMsgSelectionStage.java | 16 +-
.../stages/TestRebalancePipeline.java | 41 ++--
.../stages/TestResourceComputationStage.java | 8 +-
.../helix/healthcheck/TestAddDropAlert.java | 10 +-
.../helix/healthcheck/TestExpandAlert.java | 10 +-
.../helix/healthcheck/TestSimpleAlert.java | 9 +-
.../healthcheck/TestSimpleWildcardAlert.java | 9 +-
.../helix/healthcheck/TestStalenessAlert.java | 9 +-
.../helix/healthcheck/TestWildcardAlert.java | 9 +-
.../integration/TestCleanupExternalView.java | 18 +-
.../TestCustomizedIdealStateRebalancer.java | 8 +-
.../TestEnablePartitionDuringDisable.java | 14 +-
.../TestMessagePartitionStateMismatch.java | 39 ++--
.../helix/integration/TestMessageThrottle2.java | 25 +--
.../helix/integration/TestMessagingService.java | 34 ++--
.../TestParticipantErrorMessage.java | 13 +-
.../integration/TestResetPartitionState.java | 15 +-
.../helix/integration/TestSchedulerMessage.java | 104 +++++-----
.../TestSessionExpiryInTransition.java | 8 +-
.../helix/integration/TestZkReconnect.java | 5 +-
.../manager/TestParticipantManager.java | 4 +-
.../TestDefaultControllerMsgHandlerFactory.java | 10 +-
.../helix/manager/zk/TestZkClusterManager.java | 19 +-
.../helix/messaging/TestAsyncCallback.java | 9 +-
.../helix/messaging/TestAsyncCallbackSvc.java | 50 +++--
.../messaging/TestDefaultMessagingService.java | 4 +-
.../handling/TestHelixTaskExecutor.java | 75 ++++----
.../helix/mock/controller/MockController.java | 19 +-
.../mock/controller/MockControllerProcess.java | 17 +-
.../helix/mock/participant/DummyProcess.java | 7 +-
.../helix/mock/participant/ErrTransition.java | 10 +-
.../StoreAccessDiffNodeTransition.java | 4 +-
.../StoreAccessOneNodeTransition.java | 5 +-
.../org/apache/helix/model/TestConstraint.java | 13 +-
.../TestDistControllerStateModel.java | 22 +--
.../apache/helix/examples/BootstrapHandler.java | 7 +-
.../apache/helix/examples/BootstrapProcess.java | 2 +-
.../apache/helix/examples/DummyParticipant.java | 25 +--
.../examples/MasterSlaveStateModelFactory.java | 16 +-
.../helix/filestore/FileStoreStateModel.java | 38 ++--
.../helix/taskexecution/TaskStateModel.java | 17 +-
116 files changed, 1377 insertions(+), 1368 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/IdealStateResource.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/IdealStateResource.java b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/IdealStateResource.java
index 87e6cd7..571aff7 100644
--- a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/IdealStateResource.java
+++ b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/IdealStateResource.java
@@ -25,8 +25,8 @@ import java.util.Map;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.PropertyKey;
-import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.ZNRecord;
import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.IdealState;
import org.apache.helix.tools.ClusterSetup;
@@ -98,7 +98,7 @@ public class IdealStateResource extends Resource {
String message =
ClusterRepresentationUtil.getClusterPropertyAsString(zkClient, clusterName,
- keyBuilder.idealStates(resourceName), MediaType.APPLICATION_JSON);
+ keyBuilder.idealState(resourceName), MediaType.APPLICATION_JSON);
StringRepresentation representation =
new StringRepresentation(message, MediaType.APPLICATION_JSON);
@@ -123,7 +123,7 @@ public class IdealStateResource extends Resource {
HelixDataAccessor accessor =
ClusterRepresentationUtil.getClusterDataAccessor(zkClient, clusterName);
- accessor.setProperty(accessor.keyBuilder().idealStates(resourceName), new IdealState(
+ accessor.setProperty(accessor.keyBuilder().idealState(resourceName), new IdealState(
newIdealState));
} else if (command.equalsIgnoreCase(ClusterSetup.rebalance)) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/ResourceGroupResource.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/ResourceGroupResource.java b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/ResourceGroupResource.java
index 365b212..885b506 100644
--- a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/ResourceGroupResource.java
+++ b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/ResourceGroupResource.java
@@ -95,7 +95,7 @@ public class ResourceGroupResource extends Resource {
String message =
ClusterRepresentationUtil.getClusterPropertyAsString(zkClient, clusterName,
- keyBuilder.idealStates(resourceName), MediaType.APPLICATION_JSON);
+ keyBuilder.idealState(resourceName), MediaType.APPLICATION_JSON);
StringRepresentation representation =
new StringRepresentation(message, MediaType.APPLICATION_JSON);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/SchedulerTasksResource.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/SchedulerTasksResource.java b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/SchedulerTasksResource.java
index 2e48de4..93f5b3f 100644
--- a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/SchedulerTasksResource.java
+++ b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/SchedulerTasksResource.java
@@ -31,6 +31,7 @@ import org.apache.helix.HelixException;
import org.apache.helix.InstanceType;
import org.apache.helix.PropertyPathConfig;
import org.apache.helix.PropertyType;
+import org.apache.helix.api.Id;
import org.apache.helix.manager.zk.DefaultSchedulerMessageHandlerFactory;
import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.LiveInstance;
@@ -117,7 +118,6 @@ public class SchedulerTasksResource extends Resource {
ClusterRepresentationUtil.getClusterDataAccessor(zkClient, clusterName);
LiveInstance liveInstance =
accessor.getProperty(accessor.keyBuilder().liveInstance(instanceName));
- String sessionId = liveInstance.getSessionIdString();
StringRepresentation representation = new StringRepresentation("");// (ClusterRepresentationUtil.ObjectToJson(instanceConfigs),
// MediaType.APPLICATION_JSON);
@@ -153,12 +153,12 @@ public class SchedulerTasksResource extends Resource {
}
Message schedulerMessage =
- new Message(MessageType.SCHEDULER_MSG, UUID.randomUUID().toString());
+ new Message(MessageType.SCHEDULER_MSG, Id.message(UUID.randomUUID().toString()));
schedulerMessage.getRecord().getSimpleFields().put(CRITERIA, criteriaString);
schedulerMessage.getRecord().getMapFields().put(MESSAGETEMPLATE, messageTemplate);
- schedulerMessage.setTgtSessionId(leader.getSessionIdString());
+ schedulerMessage.setTgtSessionId(Id.session(leader.getSessionId().stringify()));
schedulerMessage.setTgtName("CONTROLLER");
schedulerMessage.setSrcInstanceType(InstanceType.CONTROLLER);
String taskQueueName =
@@ -167,22 +167,23 @@ public class SchedulerTasksResource extends Resource {
schedulerMessage.getRecord().setSimpleField(
DefaultSchedulerMessageHandlerFactory.SCHEDULER_TASK_QUEUE, taskQueueName);
}
- accessor.setProperty(accessor.keyBuilder().controllerMessage(schedulerMessage.getMsgIdString()),
+ accessor.setProperty(
+ accessor.keyBuilder().controllerMessage(schedulerMessage.getMsgId().stringify()),
schedulerMessage);
Map<String, String> resultMap = new HashMap<String, String>();
resultMap.put("StatusUpdatePath", PropertyPathConfig.getPath(
PropertyType.STATUSUPDATES_CONTROLLER, clusterName, MessageType.SCHEDULER_MSG.toString(),
- schedulerMessage.getMsgIdString()));
+ schedulerMessage.getMsgId().stringify()));
resultMap.put("MessageType", Message.MessageType.SCHEDULER_MSG.toString());
- resultMap.put("MsgId", schedulerMessage.getMsgIdString());
+ resultMap.put("MsgId", schedulerMessage.getMsgId().stringify());
// Assemble the rest URL for task status update
String ipAddress = InetAddress.getLocalHost().getCanonicalHostName();
String url =
"http://" + ipAddress + ":" + getContext().getAttributes().get(RestAdminApplication.PORT)
+ "/clusters/" + clusterName + "/Controller/statusUpdates/SCHEDULER_MSG/"
- + schedulerMessage.getMsgIdString();
+ + schedulerMessage.getMsgId();
resultMap.put("statusUpdateUrl", url);
getResponse().setEntity(ClusterRepresentationUtil.ObjectToJson(resultMap),
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-admin-webapp/src/test/java/org/apache/helix/tools/TestResetPartitionState.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/test/java/org/apache/helix/tools/TestResetPartitionState.java b/helix-admin-webapp/src/test/java/org/apache/helix/tools/TestResetPartitionState.java
index c3980bf..ec52151 100644
--- a/helix-admin-webapp/src/test/java/org/apache/helix/tools/TestResetPartitionState.java
+++ b/helix-admin-webapp/src/test/java/org/apache/helix/tools/TestResetPartitionState.java
@@ -29,6 +29,7 @@ import org.apache.helix.NotificationContext;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.TestHelper;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.State;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.mock.controller.ClusterController;
@@ -67,9 +68,9 @@ public class TestResetPartitionState extends AdminTestBase {
@Override
public void doTransition(Message message, NotificationContext context) {
super.doTransition(message, context);
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
- if (fromState.equals("ERROR") && toState.equals("OFFLINE")) {
+ State fromState = message.getFromState();
+ State toState = message.getToState();
+ if (fromState.toString().equals("ERROR") && toState.toString().equals("OFFLINE")) {
// System.err.println("doReset() invoked");
_errToOfflineInvoked.incrementAndGet();
}
@@ -191,8 +192,8 @@ public class TestResetPartitionState extends AdminTestBase {
Builder keyBuilder = accessor.keyBuilder();
LiveInstance liveInstance = accessor.getProperty(keyBuilder.liveInstance(instance));
- accessor.removeProperty(keyBuilder.stateTransitionStatus(instance, liveInstance.getSessionIdString(),
- resource, partition));
+ accessor.removeProperty(keyBuilder.stateTransitionStatus(instance, liveInstance.getSessionId()
+ .stringify(), resource, partition));
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-agent/src/main/java/org/apache/helix/agent/AgentStateModel.java
----------------------------------------------------------------------
diff --git a/helix-agent/src/main/java/org/apache/helix/agent/AgentStateModel.java b/helix-agent/src/main/java/org/apache/helix/agent/AgentStateModel.java
index 393da23..b0106a0 100644
--- a/helix-agent/src/main/java/org/apache/helix/agent/AgentStateModel.java
+++ b/helix-agent/src/main/java/org/apache/helix/agent/AgentStateModel.java
@@ -29,9 +29,10 @@ import java.util.regex.Pattern;
import org.apache.helix.ExternalCommand;
import org.apache.helix.HelixManager;
import org.apache.helix.NotificationContext;
+import org.apache.helix.api.State;
import org.apache.helix.model.HelixConfigScope;
-import org.apache.helix.model.Message;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
+import org.apache.helix.model.Message;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.helix.participant.statemachine.StateModel;
import org.apache.helix.participant.statemachine.StateModelInfo;
@@ -71,21 +72,24 @@ public class AgentStateModel extends StateModel {
HelixManager manager = context.getManager();
String clusterName = manager.getClusterName();
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
// construct keys for command-config
- String cmdKey = buildKey(fromState, toState, CommandAttribute.COMMAND);
- String workingDirKey = buildKey(fromState, toState, CommandAttribute.WORKING_DIR);
- String timeoutKey = buildKey(fromState, toState, CommandAttribute.TIMEOUT);
- String pidFileKey = buildKey(fromState, toState, CommandAttribute.PID_FILE);
+ String cmdKey = buildKey(fromState.toString(), toState.toString(), CommandAttribute.COMMAND);
+ String workingDirKey =
+ buildKey(fromState.toString(), toState.toString(), CommandAttribute.WORKING_DIR);
+ String timeoutKey =
+ buildKey(fromState.toString(), toState.toString(), CommandAttribute.TIMEOUT);
+ String pidFileKey =
+ buildKey(fromState.toString(), toState.toString(), CommandAttribute.PID_FILE);
List<String> cmdConfigKeys = Arrays.asList(cmdKey, workingDirKey, timeoutKey, pidFileKey);
// read command from resource-scope configures
if (cmd == null) {
HelixConfigScope resourceScope =
new HelixConfigScopeBuilder(ConfigScopeProperty.RESOURCE).forCluster(clusterName)
- .forResource(message.getResourceName()).build();
+ .forResource(message.getResourceId().stringify()).build();
Map<String, String> cmdKeyValueMap =
manager.getConfigAccessor().get(resourceScope, cmdConfigKeys);
if (cmdKeyValueMap != null) {
@@ -112,8 +116,8 @@ public class AgentStateModel extends StateModel {
}
if (cmd == null) {
- throw new Exception("Unable to find command for transition from:" + message.getFromStateString()
- + " to:" + message.getToStateString());
+ throw new Exception("Unable to find command for transition from:" + message.getFromState()
+ + " to:" + message.getToState());
}
_logger.info("Executing command: " + cmd + ", using workingDir: " + workingDir + ", timeout: "
+ timeout + ", on " + manager.getInstanceName());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/Controller.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/Controller.java b/helix-core/src/main/java/org/apache/helix/api/Controller.java
index d056e13..c47d603 100644
--- a/helix-core/src/main/java/org/apache/helix/api/Controller.java
+++ b/helix-core/src/main/java/org/apache/helix/api/Controller.java
@@ -38,8 +38,8 @@ public class Controller {
if (liveInstance != null) {
_runningInstance =
- new RunningInstance(new SessionId(liveInstance.getSessionIdString()), new HelixVersion(
- liveInstance.getHelixVersionString()), new ProcId(liveInstance.getLiveInstance()));
+ new RunningInstance(liveInstance.getSessionId(), liveInstance.getHelixVersion(),
+ liveInstance.getProcessId());
} else {
_runningInstance = null;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/CurState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/CurState.java b/helix-core/src/main/java/org/apache/helix/api/CurState.java
deleted file mode 100644
index b6f64a9..0000000
--- a/helix-core/src/main/java/org/apache/helix/api/CurState.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package org.apache.helix.api;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.helix.model.CurrentState;
-
-import com.google.common.collect.ImmutableMap;
-
-/**
- * Current state per participant per resource
- */
-public class CurState {
- private final ResourceId _resourceId;
- private final ParticipantId _participantId;
-
- /**
- * map of partition-id to state
- */
- final Map<PartitionId, State> _stateMap;
-
- /**
- * Construct current state
- * @param resource
- * @param participant
- * @param currentState
- */
- public CurState(ResourceId resourceId, ParticipantId participantId, CurrentState currentState) {
- _resourceId = resourceId;
- _participantId = participantId;
-
- Map<PartitionId, State> stateMap = new HashMap<PartitionId, State>();
- Map<String, String> currentStateMap = currentState.getPartitionStateStringMap();
- for (String partitionId : currentStateMap.keySet()) {
- String state = currentStateMap.get(partitionId);
- stateMap.put(new PartitionId(resourceId, PartitionId.stripResourceId(partitionId)),
- new State(state));
- }
- _stateMap = ImmutableMap.copyOf(stateMap);
- }
-
- /**
- * Get current state for a partition
- * @param partition-id
- * @return state of the partition or null if partition not exist
- */
- public State getState(PartitionId partitionId) {
- return _stateMap.get(partitionId);
- }
-
- /**
- * Get the set of partition-id's in the current state
- * @return set of partition-id's or empty set if none
- */
- public Set<PartitionId> getPartitionIdSet() {
- return _stateMap.keySet();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/ExtView.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/ExtView.java b/helix-core/src/main/java/org/apache/helix/api/ExtView.java
deleted file mode 100644
index 3a5d4ac..0000000
--- a/helix-core/src/main/java/org/apache/helix/api/ExtView.java
+++ /dev/null
@@ -1,62 +0,0 @@
-package org.apache.helix.api;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.Map;
-
-import com.google.common.collect.ImmutableMap;
-
-/**
- * External view of a resource
- */
-public class ExtView {
- private final ResourceId _resourceId;
-
- /**
- * map of partition-id to map of participant-id to state
- */
- private final RscAssignment _extView;
-
- /**
- * Construct external view
- * @param stateMap map of partition-id to map of participant-id to state
- */
- public ExtView(ResourceId resourceId, Map<String, Map<String, String>> stateMap) {
- _resourceId = resourceId;
-
- // TODO convert to external view
- _extView = null;
- }
-
- /**
- * Get the state of a partition for a participant
- * @param partitionId
- * @param participantIds
- * @return the state or null if not exist
- */
- public State getState(PartitionId partitionId, ParticipantId participantId) {
- Map<ParticipantId, State> participantStateMap = _extView.getParticipantStateMap(partitionId);
- if (participantStateMap != null) {
- return participantStateMap.get(participantId);
- }
- return null;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/HelixVersion.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/HelixVersion.java b/helix-core/src/main/java/org/apache/helix/api/HelixVersion.java
index a32c9c7..b50031d 100644
--- a/helix-core/src/main/java/org/apache/helix/api/HelixVersion.java
+++ b/helix-core/src/main/java/org/apache/helix/api/HelixVersion.java
@@ -60,6 +60,9 @@ public class HelixVersion {
* @return HelixVersion
*/
public static HelixVersion from(String version) {
+ if (version == null) {
+ return null;
+ }
return new HelixVersion(version);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/Id.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/Id.java b/helix-core/src/main/java/org/apache/helix/api/Id.java
index d31084a..338ad11 100644
--- a/helix-core/src/main/java/org/apache/helix/api/Id.java
+++ b/helix-core/src/main/java/org/apache/helix/api/Id.java
@@ -32,8 +32,10 @@ public abstract class Id implements Comparable<Id> {
@Override
public boolean equals(Object that) {
- if (that instanceof ClusterId) {
+ if (that instanceof Id) {
return this.stringify().equals(((Id) that).stringify());
+ } else if (that instanceof String) {
+ return this.stringify().equals((String) that);
}
return false;
}
@@ -77,6 +79,16 @@ public abstract class Id implements Comparable<Id> {
}
/**
+ * Get a concrete partition id
+ * @param resourceId resource identifier
+ * @param partitionSuffix partition identifier relative to a resource
+ * @return PartitionId
+ */
+ public static PartitionId partition(ResourceId resourceId, String partitionSuffix) {
+ return new PartitionId(resourceId, partitionSuffix);
+ }
+
+ /**
* Get a concrete participant id
* @param participantId string participant identifier
* @return ParticipantId
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/Msg.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/Msg.java b/helix-core/src/main/java/org/apache/helix/api/Msg.java
deleted file mode 100644
index 229d742..0000000
--- a/helix-core/src/main/java/org/apache/helix/api/Msg.java
+++ /dev/null
@@ -1,68 +0,0 @@
-package org.apache.helix.api;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import org.apache.helix.model.Message;
-
-/**
- * Helix message
- */
-public class Msg {
- private final MessageId _id;
- private final SessionId _srcSessionId;
- private final SessionId _tgtSessionid;
-
- // TODO add other message fields
-
- /**
- * Construct a message
- * @param message
- */
- public Msg(Message message) {
- _id = new MessageId(message.getId());
- _srcSessionId = new SessionId(message.getSrcSessionIdString());
- _tgtSessionid = new SessionId(message.getTgtSessionIdString());
- }
-
- /**
- * Get message id
- * @return message id
- */
- public MessageId getId() {
- return _id;
- }
-
- /**
- * Get sender session id
- * @return sender session id
- */
- public SessionId getSrcSessionId() {
- return _srcSessionId;
- }
-
- /**
- * Get receiver session id
- * @return receiver session id
- */
- public SessionId getTgtSessionId() {
- return _tgtSessionid;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/ParticipantAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/ParticipantAccessor.java b/helix-core/src/main/java/org/apache/helix/api/ParticipantAccessor.java
index 58d18df..4bc502e 100644
--- a/helix-core/src/main/java/org/apache/helix/api/ParticipantAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/api/ParticipantAccessor.java
@@ -95,12 +95,12 @@ public class ParticipantAccessor {
* @param participantId
* @param msgMap map of message-id to message
*/
- public void sendMessagesToParticipant(ParticipantId participantId, Map<MessageId, Message> msgMap) {
- String participantName = participantId.stringify();
+ public void insertMessagesToParticipant(ParticipantId participantId,
+ Map<MessageId, Message> msgMap) {
List<PropertyKey> msgKeys = new ArrayList<PropertyKey>();
List<Message> msgs = new ArrayList<Message>();
for (MessageId msgId : msgMap.keySet()) {
- msgKeys.add(_keyBuilder.message(participantName, msgId.stringify()));
+ msgKeys.add(_keyBuilder.message(participantId.stringify(), msgId.stringify()));
msgs.add(msgMap.get(msgId));
}
@@ -302,8 +302,8 @@ public class ParticipantAccessor {
RunningInstance runningInstance = null;
if (liveInstance != null) {
runningInstance =
- new RunningInstance(new SessionId(liveInstance.getSessionIdString()), new HelixVersion(
- liveInstance.getHelixVersionString()), new ProcId(liveInstance.getLiveInstance()));
+ new RunningInstance(liveInstance.getSessionId(), liveInstance.getHelixVersion(),
+ liveInstance.getProcessId());
}
Map<MessageId, Message> msgMap = new HashMap<MessageId, Message>();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/PartitionId.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/PartitionId.java b/helix-core/src/main/java/org/apache/helix/api/PartitionId.java
index 04dbab1..9257eee 100644
--- a/helix-core/src/main/java/org/apache/helix/api/PartitionId.java
+++ b/helix-core/src/main/java/org/apache/helix/api/PartitionId.java
@@ -30,6 +30,10 @@ public class PartitionId extends Id {
@Override
public String stringify() {
+ // check in case the partition name is instantiated incorrectly
+ if (_resourceId.stringify().equals(_partitionName)) {
+ return _partitionName;
+ }
return String.format("%s_%s", _resourceId, _partitionName);
}
@@ -38,6 +42,9 @@ public class PartitionId extends Id {
* @return
*/
public static String stripResourceId(String partitionName) {
+ if (partitionName == null || !partitionName.contains("_")) {
+ return partitionName;
+ }
return partitionName.substring(partitionName.lastIndexOf("_") + 1);
}
@@ -46,6 +53,9 @@ public class PartitionId extends Id {
* @return
*/
public static ResourceId extractResourceId(String partitionName) {
+ if (partitionName == null || !partitionName.contains("_")) {
+ return new ResourceId(partitionName);
+ }
return new ResourceId(partitionName.substring(0, partitionName.lastIndexOf("_")));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/RebalancerConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/RebalancerConfig.java b/helix-core/src/main/java/org/apache/helix/api/RebalancerConfig.java
index cf4fbbc..34498cd 100644
--- a/helix-core/src/main/java/org/apache/helix/api/RebalancerConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/api/RebalancerConfig.java
@@ -20,16 +20,17 @@ package org.apache.helix.api;
*/
import org.apache.helix.model.IdealState.RebalanceMode;
+import org.apache.helix.model.ResourceAssignment;
public class RebalancerConfig {
private final RebalanceMode _rebalancerMode;
private final RebalancerRef _rebalancerRef;
private final StateModelDefId _stateModelDefId;
- private final RscAssignment _resourceAssignment;
+ private final ResourceAssignment _resourceAssignment;
public RebalancerConfig(RebalanceMode mode, RebalancerRef rebalancerRef,
- StateModelDefId stateModelDefId, RscAssignment resourceAssignment) {
+ StateModelDefId stateModelDefId, ResourceAssignment resourceAssignment) {
_rebalancerMode = mode;
_rebalancerRef = rebalancerRef;
_stateModelDefId = stateModelDefId;
@@ -64,7 +65,7 @@ public class RebalancerConfig {
* Get the ideal node and state assignment of the resource
* @return resource assignment
*/
- public RscAssignment getResourceAssignment() {
+ public ResourceAssignment getResourceAssignment() {
return _resourceAssignment;
}
@@ -75,7 +76,7 @@ public class RebalancerConfig {
private RebalanceMode _mode = RebalanceMode.NONE;
private RebalancerRef _rebalancerRef;
private StateModelDefId _stateModelDefId;
- private RscAssignment _resourceAssignment;
+ private ResourceAssignment _resourceAssignment;
/**
* Set the rebalancer mode
@@ -111,7 +112,7 @@ public class RebalancerConfig {
* @param resourceAssignment resource assignment
* @return Builder
*/
- public Builder resourceAssignment(RscAssignment resourceAssignment) {
+ public Builder resourceAssignment(ResourceAssignment resourceAssignment) {
_resourceAssignment = resourceAssignment;
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/RebalancerRef.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/RebalancerRef.java b/helix-core/src/main/java/org/apache/helix/api/RebalancerRef.java
index 033e1e7..7f33be7 100644
--- a/helix-core/src/main/java/org/apache/helix/api/RebalancerRef.java
+++ b/helix-core/src/main/java/org/apache/helix/api/RebalancerRef.java
@@ -57,6 +57,8 @@ public class RebalancerRef {
public boolean equals(Object that) {
if (that instanceof RebalancerRef) {
return this.toString().equals(((RebalancerRef) that).toString());
+ } else if (that instanceof String) {
+ return this.toString().equals((String) that);
}
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/Resource.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/Resource.java b/helix-core/src/main/java/org/apache/helix/api/Resource.java
index 354d0a1..2c9ba8a 100644
--- a/helix-core/src/main/java/org/apache/helix/api/Resource.java
+++ b/helix-core/src/main/java/org/apache/helix/api/Resource.java
@@ -22,6 +22,7 @@ package org.apache.helix.api;
import java.util.HashSet;
import java.util.Set;
+import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.ResourceAssignment;
@@ -36,7 +37,7 @@ public class Resource {
private final Set<Partition> _partitionSet;
- private final ExtView _externalView;
+ private final ExternalView _externalView;
// TODO move construct logic to ResourceAccessor
/**
@@ -71,7 +72,7 @@ public class Resource {
* @param externalView external view of the resource
* @param rebalancerConfig configuration properties for rebalancing this resource
*/
- public Resource(ResourceId id, Set<Partition> partitionSet, ExtView externalView,
+ public Resource(ResourceId id, Set<Partition> partitionSet, ExternalView externalView,
RebalancerConfig rebalancerConfig) {
_id = id;
_partitionSet = ImmutableSet.copyOf(partitionSet);
@@ -91,7 +92,7 @@ public class Resource {
* Get the external view of the resource
* @return the external view of the resource
*/
- public ExtView getExternalView() {
+ public ExternalView getExternalView() {
return _externalView;
}
@@ -109,7 +110,7 @@ public class Resource {
public static class Builder {
private final ResourceId _id;
private final Set<Partition> _partitionSet;
- private ExtView _externalView;
+ private ExternalView _externalView;
private RebalancerConfig _rebalancerConfig;
/**
@@ -136,7 +137,7 @@ public class Resource {
* @param extView currently served replica placement and state
* @return Builder
*/
- public Builder externalView(ExtView extView) {
+ public Builder externalView(ExternalView extView) {
_externalView = extView;
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/ResourceAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/ResourceAccessor.java b/helix-core/src/main/java/org/apache/helix/api/ResourceAccessor.java
index a3d2543..ba7cd3d 100644
--- a/helix-core/src/main/java/org/apache/helix/api/ResourceAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/api/ResourceAccessor.java
@@ -23,6 +23,7 @@ import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
+import org.apache.helix.model.ResourceAssignment;
public class ResourceAccessor {
private final ClusterId _clusterId;
@@ -40,8 +41,7 @@ public class ResourceAccessor {
* @param resourceId
* @param resourceAssignment
*/
- public void setResourceAssignment(ResourceId resourceId, RscAssignment resourceAssignment) {
- // TODO impl this
+ public void setRresourceAssignment(ResourceId resourceId, ResourceAssignment resourceAssignment) {
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/RscAssignment.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/RscAssignment.java b/helix-core/src/main/java/org/apache/helix/api/RscAssignment.java
deleted file mode 100644
index 90b77e3..0000000
--- a/helix-core/src/main/java/org/apache/helix/api/RscAssignment.java
+++ /dev/null
@@ -1,140 +0,0 @@
-package org.apache.helix.api;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.helix.model.ResourceAssignment;
-
-import com.google.common.collect.ImmutableMap;
-
-public class RscAssignment {
- private final Map<PartitionId, Map<ParticipantId, State>> _resourceAssignment;
-
- /**
- * Construct an assignment from a physically-stored assignment
- * @param rscAssignment the assignment
- */
- public RscAssignment(ResourceAssignment rscAssignment) {
- Map<PartitionId, Map<ParticipantId, State>> resourceAssignment =
- new HashMap<PartitionId, Map<ParticipantId, State>>();
-
- for (org.apache.helix.model.Partition partition : rscAssignment.getMappedPartitions()) {
- Map<ParticipantId, State> replicaMap = new HashMap<ParticipantId, State>();
- Map<String, String> rawReplicaMap = rscAssignment.getReplicaMap(partition);
- for (String participantId : rawReplicaMap.keySet()) {
- replicaMap.put(new ParticipantId(participantId),
- new State(rawReplicaMap.get(participantId)));
- }
- resourceAssignment.put(new PartitionId(new ResourceId(rscAssignment.getResourceName()),
- partition.getPartitionName()), replicaMap);
- }
-
- _resourceAssignment = ImmutableMap.copyOf(resourceAssignment);
- }
-
- /**
- * Build an assignment from a map of assigned replicas
- * @param resourceAssignment map of (partition, participant, state)
- */
- public RscAssignment(Map<PartitionId, Map<ParticipantId, State>> resourceAssignment) {
- ImmutableMap.Builder<PartitionId, Map<ParticipantId, State>> mapBuilder =
- new ImmutableMap.Builder<PartitionId, Map<ParticipantId, State>>();
- for (PartitionId partitionId : resourceAssignment.keySet()) {
- mapBuilder.put(partitionId, ImmutableMap.copyOf(resourceAssignment.get(partitionId)));
- }
- _resourceAssignment = mapBuilder.build();
- }
-
- /**
- * Get the partitions currently with assignments
- * @return set of partition ids
- */
- public Set<PartitionId> getAssignedPartitions() {
- return _resourceAssignment.keySet();
- }
-
- /**
- * Get the replica assignment map for a partition
- * @param partitionId the partition to look up
- * @return map of (participant id, state)
- */
- public Map<ParticipantId, State> getParticipantStateMap(PartitionId partitionId) {
- return _resourceAssignment.get(partitionId);
- }
-
- /**
- * Assemble a full assignment
- */
- public static class Builder {
- private final Map<PartitionId, Map<ParticipantId, State>> _resourceAssignment;
-
- /**
- * Instantiate the builder
- */
- public Builder() {
- _resourceAssignment = new HashMap<PartitionId, Map<ParticipantId, State>>();
- }
-
- /**
- * Add assignments for a partition
- * @param partitionId partition to assign
- * @param replicaMap map of participant and state for each replica
- * @return Builder
- */
- public Builder addAssignments(PartitionId partitionId, Map<ParticipantId, State> replicaMap) {
- if (!_resourceAssignment.containsKey(partitionId)) {
- _resourceAssignment.put(partitionId, replicaMap);
- } else {
- _resourceAssignment.get(partitionId).putAll(replicaMap);
- }
- return this;
- }
-
- /**
- * Assign a single replica
- * @param partitionId partition to assign
- * @param participantId participant to host the replica
- * @param state replica state
- * @return Builder
- */
- public Builder addAssignment(PartitionId partitionId, ParticipantId participantId, State state) {
- Map<ParticipantId, State> replicaMap;
- if (!_resourceAssignment.containsKey(partitionId)) {
- replicaMap = new HashMap<ParticipantId, State>();
- _resourceAssignment.put(partitionId, replicaMap);
- } else {
- replicaMap = _resourceAssignment.get(partitionId);
- }
- replicaMap.put(participantId, state);
- return this;
- }
-
- /**
- * Build the resource assignment
- * @return instantiated RscAssignment
- */
- public RscAssignment build() {
- return new RscAssignment(_resourceAssignment);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/api/State.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/State.java b/helix-core/src/main/java/org/apache/helix/api/State.java
index 0fc8bc7..2a1f993 100644
--- a/helix-core/src/main/java/org/apache/helix/api/State.java
+++ b/helix-core/src/main/java/org/apache/helix/api/State.java
@@ -38,10 +38,17 @@ public class State {
public boolean equals(Object that) {
if (that instanceof State) {
return this.toString().equals(((State) that).toString());
+ } else if (that instanceof String) {
+ return _state.equals((String) that);
}
return false;
}
+ @Override
+ public int hashCode() {
+ return _state.hashCode();
+ }
+
/**
* Get a State from a state name
* @param state state name
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index 5ba4362..7d69152 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -480,7 +480,7 @@ public class GenericHelixController implements ConfigChangeListener, IdealStateC
Map<String, LiveInstance> curSessions = new HashMap<String, LiveInstance>();
for (LiveInstance liveInstance : liveInstances) {
curInstances.put(liveInstance.getInstanceName(), liveInstance);
- curSessions.put(liveInstance.getSessionIdString(), liveInstance);
+ curSessions.put(liveInstance.getSessionId().stringify(), liveInstance);
}
Map<String, LiveInstance> lastInstances = _lastSeenInstances.get();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
index 6d64bb8..b796b72 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/AutoRebalancer.java
@@ -29,6 +29,7 @@ import java.util.Set;
import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
import org.apache.helix.controller.rebalancer.util.ConstraintBasedAssignment;
import org.apache.helix.controller.stages.ClusterDataCache;
import org.apache.helix.controller.stages.CurrentStateOutput;
@@ -134,7 +135,8 @@ public class AutoRebalancer implements Rebalancer {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing resource:" + resource.getResourceName());
}
- ResourceAssignment partitionMapping = new ResourceAssignment(resource.getResourceName());
+ ResourceAssignment partitionMapping =
+ new ResourceAssignment(Id.resource(resource.getResourceName()));
for (String partitionName : partitions) {
Partition partition = new Partition(partitionName);
Map<String, String> currentStateMap =
@@ -147,7 +149,8 @@ public class AutoRebalancer implements Rebalancer {
Map<String, String> bestStateForPartition =
ConstraintBasedAssignment.computeAutoBestStateForPartition(clusterData, stateModelDef,
preferenceList, currentStateMap, disabledInstancesForPartition);
- partitionMapping.addReplicaMap(partition, bestStateForPartition);
+ partitionMapping.addReplicaMap(Id.partition(partitionName),
+ ResourceAssignment.replicaMapFromStringMap(bestStateForPartition));
}
return partitionMapping;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
index 5de420c..b22562e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/CustomRebalancer.java
@@ -25,6 +25,7 @@ import java.util.Set;
import org.apache.helix.HelixDefinedState;
import org.apache.helix.HelixManager;
+import org.apache.helix.api.Id;
import org.apache.helix.controller.stages.ClusterDataCache;
import org.apache.helix.controller.stages.CurrentStateOutput;
import org.apache.helix.model.IdealState;
@@ -60,7 +61,8 @@ public class CustomRebalancer implements Rebalancer {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing resource:" + resource.getResourceName());
}
- ResourceAssignment partitionMapping = new ResourceAssignment(resource.getResourceName());
+ ResourceAssignment partitionMapping =
+ new ResourceAssignment(Id.resource(resource.getResourceName()));
for (Partition partition : resource.getPartitions()) {
Map<String, String> currentStateMap =
currentStateOutput.getCurrentStateMap(resource.getResourceName(), partition);
@@ -71,7 +73,8 @@ public class CustomRebalancer implements Rebalancer {
Map<String, String> bestStateForPartition =
computeCustomizedBestStateForPartition(clusterData, stateModelDef, idealStateMap,
currentStateMap, disabledInstancesForPartition);
- partitionMapping.addReplicaMap(partition, bestStateForPartition);
+ partitionMapping.addReplicaMap(Id.partition(partition.getPartitionName()),
+ ResourceAssignment.replicaMapFromStringMap(bestStateForPartition));
}
return partitionMapping;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/rebalancer/SemiAutoRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/SemiAutoRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/SemiAutoRebalancer.java
index b096817..a488548 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/SemiAutoRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/SemiAutoRebalancer.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.helix.HelixManager;
+import org.apache.helix.api.Id;
import org.apache.helix.controller.rebalancer.util.ConstraintBasedAssignment;
import org.apache.helix.controller.stages.ClusterDataCache;
import org.apache.helix.controller.stages.CurrentStateOutput;
@@ -59,7 +60,8 @@ public class SemiAutoRebalancer implements Rebalancer {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing resource:" + resource.getResourceName());
}
- ResourceAssignment partitionMapping = new ResourceAssignment(resource.getResourceName());
+ ResourceAssignment partitionMapping =
+ new ResourceAssignment(Id.resource(resource.getResourceName()));
for (Partition partition : resource.getPartitions()) {
Map<String, String> currentStateMap =
currentStateOutput.getCurrentStateMap(resource.getResourceName(), partition);
@@ -71,7 +73,8 @@ public class SemiAutoRebalancer implements Rebalancer {
Map<String, String> bestStateForPartition =
ConstraintBasedAssignment.computeAutoBestStateForPartition(clusterData, stateModelDef,
preferenceList, currentStateMap, disabledInstancesForPartition);
- partitionMapping.addReplicaMap(partition, bestStateForPartition);
+ partitionMapping.addReplicaMap(Id.partition(partition.getPartitionName()),
+ ResourceAssignment.replicaMapFromStringMap(bestStateForPartition));
}
return partitionMapping;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
index e812e16..10240db 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java
@@ -22,6 +22,9 @@ package org.apache.helix.controller.stages;
import java.util.Map;
import org.apache.helix.HelixManager;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.ParticipantId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.controller.rebalancer.AutoRebalancer;
@@ -119,8 +122,9 @@ public class BestPossibleStateCalcStage extends AbstractBaseStage {
ResourceAssignment partitionStateAssignment =
rebalancer.computeResourceMapping(resource, idealState, currentStateOutput, cache);
for (Partition partition : resource.getPartitions()) {
- Map<String, String> newStateMap = partitionStateAssignment.getReplicaMap(partition);
- output.setState(resourceName, partition, newStateMap);
+ Map<ParticipantId, State> newStateMap =
+ partitionStateAssignment.getReplicaMap(Id.partition(partition.getPartitionName()));
+ output.setParticipantStateMap(resourceName, partition, newStateMap);
}
}
return output;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateOutput.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateOutput.java b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateOutput.java
index 3da9bef..fecefd5 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateOutput.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateOutput.java
@@ -23,6 +23,8 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.apache.helix.api.ParticipantId;
+import org.apache.helix.api.State;
import org.apache.helix.model.Partition;
public class BestPossibleStateOutput {
@@ -42,6 +44,16 @@ public class BestPossibleStateOutput {
map.put(resource, bestInstanceStateMappingForResource);
}
+ public void setParticipantStateMap(String resourceName, Partition partition,
+ Map<ParticipantId, State> bestInstanceStateMappingForResource) {
+ Map<String, String> rawStateMap = new HashMap<String, String>();
+ for (ParticipantId participantId : bestInstanceStateMappingForResource.keySet()) {
+ rawStateMap.put(participantId.stringify(),
+ bestInstanceStateMappingForResource.get(participantId).toString());
+ }
+ setState(resourceName, partition, rawStateMap);
+ }
+
public Map<String, String> getInstanceStateMap(String resourceName, Partition resource) {
Map<Partition, Map<String, String>> map = _dataMap.get(resourceName);
if (map != null) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
index 062f468..f50b95c 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
@@ -25,17 +25,17 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixConstants.StateModelToken;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.model.ClusterConstraints;
+import org.apache.helix.model.ClusterConstraints.ConstraintType;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
import org.apache.helix.model.StateModelDefinition;
-import org.apache.helix.model.ClusterConstraints.ConstraintType;
import org.apache.log4j.Logger;
/**
@@ -72,7 +72,7 @@ public class ClusterDataCache {
_liveInstanceMap = accessor.getChildValuesMap(keyBuilder.liveInstances());
for (LiveInstance instance : _liveInstanceMap.values()) {
- LOG.trace("live instance: " + instance.getInstanceName() + " " + instance.getSessionIdString());
+ LOG.trace("live instance: " + instance.getParticipantId() + " " + instance.getSessionId());
}
_stateModelDefMap = accessor.getChildValuesMap(keyBuilder.stateModelDefs());
@@ -90,7 +90,7 @@ public class ClusterDataCache {
new HashMap<String, Map<String, Map<String, CurrentState>>>();
for (String instanceName : _liveInstanceMap.keySet()) {
LiveInstance liveInstance = _liveInstanceMap.get(instanceName);
- String sessionId = liveInstance.getSessionIdString();
+ String sessionId = liveInstance.getSessionId().stringify();
if (!allCurStateMap.containsKey(instanceName)) {
allCurStateMap.put(instanceName, new HashMap<String, Map<String, CurrentState>>());
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
index a1b7b0d..c4fb496 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CompatibilityCheckStage.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
+import org.apache.helix.api.HelixVersion;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.LiveInstance;
@@ -46,7 +47,8 @@ public class CompatibilityCheckStage extends AbstractBaseStage {
HelixManagerProperties properties = manager.getProperties();
Map<String, LiveInstance> liveInstanceMap = cache.getLiveInstances();
for (LiveInstance liveInstance : liveInstanceMap.values()) {
- String participantVersion = liveInstance.getHelixVersionString();
+ HelixVersion version = liveInstance.getHelixVersion();
+ String participantVersion = (version != null) ? version.toString() : null;
if (!properties.isParticipantCompatible(participantVersion)) {
String errorMsg =
"incompatible participant. pipeline will not continue. " + "controller: "
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
index 1e4e72b..4499cd0 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
@@ -22,14 +22,16 @@ package org.apache.helix.controller.stages;
import java.util.List;
import java.util.Map;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.ResourceId;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
+import org.apache.helix.model.Message.MessageType;
import org.apache.helix.model.Partition;
import org.apache.helix.model.Resource;
-import org.apache.helix.model.Message.MessageType;
/**
* For each LiveInstances select currentState and message whose sessionId matches
@@ -57,32 +59,32 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
if (!MessageType.STATE_TRANSITION.toString().equalsIgnoreCase(message.getMsgType())) {
continue;
}
- if (!instance.getSessionIdString().equals(message.getTgtSessionIdString())) {
+ if (!instance.getSessionId().equals(message.getTgtSessionId())) {
continue;
}
- String resourceName = message.getResourceName();
- Resource resource = resourceMap.get(resourceName);
+ ResourceId resourceId = message.getResourceId();
+ Resource resource = resourceMap.get(resourceId.stringify());
if (resource == null) {
continue;
}
if (!message.getBatchMessageMode()) {
- String partitionName = message.getPartitionName();
- Partition partition = resource.getPartition(partitionName);
+ PartitionId partitionId = message.getPartitionId();
+ Partition partition = resource.getPartition(partitionId.stringify());
if (partition != null) {
- currentStateOutput.setPendingState(resourceName, partition, instanceName,
- message.getToStateString());
+ currentStateOutput.setPendingState(resourceId.stringify(), partition, instanceName,
+ message.getToState().toString());
} else {
// log
}
} else {
- List<String> partitionNames = message.getPartitionNames();
+ List<PartitionId> partitionNames = message.getPartitionIds();
if (!partitionNames.isEmpty()) {
- for (String partitionName : partitionNames) {
- Partition partition = resource.getPartition(partitionName);
+ for (PartitionId partitionId : partitionNames) {
+ Partition partition = resource.getPartition(partitionId.stringify());
if (partition != null) {
- currentStateOutput.setPendingState(resourceName, partition, instanceName,
- message.getToStateString());
+ currentStateOutput.setPendingState(resourceId.stringify(), partition, instanceName,
+ message.getToState().toString());
} else {
// log
}
@@ -94,12 +96,12 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
for (LiveInstance instance : liveInstances.values()) {
String instanceName = instance.getInstanceName();
- String clientSessionId = instance.getSessionIdString();
+ String clientSessionId = instance.getSessionId().stringify();
Map<String, CurrentState> currentStateMap =
cache.getCurrentState(instanceName, clientSessionId);
for (CurrentState currentState : currentStateMap.values()) {
- if (!instance.getSessionIdString().equals(currentState.getSessionIdString())) {
+ if (!instance.getSessionId().equals(currentState.getSessionId())) {
continue;
}
String resourceName = currentState.getResourceName();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
index d0237d7..0f17cf0 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
@@ -26,17 +26,20 @@ import java.util.Map;
import java.util.UUID;
import org.apache.helix.HelixManager;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.MessageId;
+import org.apache.helix.api.State;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.manager.zk.DefaultSchedulerMessageHandlerFactory;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.LiveInstance;
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.Partition;
import org.apache.helix.model.Resource;
import org.apache.helix.model.StateModelDefinition;
-import org.apache.helix.model.Message.MessageState;
-import org.apache.helix.model.Message.MessageType;
import org.apache.log4j.Logger;
/**
@@ -64,7 +67,7 @@ public class MessageGenerationPhase extends AbstractBaseStage {
Map<String, String> sessionIdMap = new HashMap<String, String>();
for (LiveInstance liveInstance : liveInstances.values()) {
- sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionIdString());
+ sessionIdMap.put(liveInstance.getInstanceName(), liveInstance.getSessionId().stringify());
}
MessageGenerationOutput output = new MessageGenerationOutput();
@@ -187,18 +190,18 @@ public class MessageGenerationPhase extends AbstractBaseStage {
private Message createMessage(HelixManager manager, String resourceName, String partitionName,
String instanceName, String currentState, String nextState, String sessionId,
String stateModelDefName, String stateModelFactoryName, int bucketSize) {
- String uuid = UUID.randomUUID().toString();
+ MessageId uuid = Id.message(UUID.randomUUID().toString());
Message message = new Message(MessageType.STATE_TRANSITION, uuid);
message.setSrcName(manager.getInstanceName());
message.setTgtName(instanceName);
message.setMsgState(MessageState.NEW);
- message.setPartitionName(partitionName);
- message.setResourceName(resourceName);
- message.setFromState(currentState);
- message.setToState(nextState);
- message.setTgtSessionId(sessionId);
- message.setSrcSessionId(manager.getSessionId());
- message.setStateModelDef(stateModelDefName);
+ message.setPartitionId(Id.partition(partitionName));
+ message.setResourceId(Id.resource(resourceName));
+ message.setFromState(State.from(currentState));
+ message.setToState(State.from(nextState));
+ message.setTgtSessionId(Id.session(sessionId));
+ message.setSrcSessionId(Id.session(manager.getSessionId()));
+ message.setStateModelDef(Id.stateModelDef(stateModelDefName));
message.setStateModelFactoryName(stateModelFactoryName);
message.setBucketSize(bucketSize);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
index 8478db3..23efacb 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import org.apache.helix.api.State;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.IdealState;
@@ -169,8 +170,8 @@ public class MessageSelectionStage extends AbstractBaseStage {
Map<Integer, List<Message>> messagesGroupByStateTransitPriority =
new TreeMap<Integer, List<Message>>();
for (Message message : messages) {
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
String transition = fromState + "-" + toState;
int priority = Integer.MAX_VALUE;
@@ -187,8 +188,8 @@ public class MessageSelectionStage extends AbstractBaseStage {
// select messages
for (List<Message> messageList : messagesGroupByStateTransitPriority.values()) {
for (Message message : messageList) {
- String fromState = message.getFromStateString();
- String toState = message.getToStateString();
+ State fromState = message.getFromState();
+ State toState = message.getToState();
if (!bounds.containsKey(fromState)) {
LOG.error("Message's fromState is not in currentState. message: " + message);
@@ -196,7 +197,7 @@ public class MessageSelectionStage extends AbstractBaseStage {
}
if (!bounds.containsKey(toState)) {
- bounds.put(toState, new Bounds(0, 0));
+ bounds.put(toState.toString(), new Bounds(0, 0));
}
// check lower bound of fromState
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
index b67a426..9818035 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java
@@ -74,7 +74,7 @@ public class ResourceComputationStage extends AbstractBaseStage {
if (availableInstances != null && availableInstances.size() > 0) {
for (LiveInstance instance : availableInstances.values()) {
String instanceName = instance.getInstanceName();
- String clientSessionId = instance.getSessionIdString();
+ String clientSessionId = instance.getSessionId().stringify();
Map<String, CurrentState> currentStateMap =
cache.getCurrentState(instanceName, clientSessionId);
@@ -98,8 +98,8 @@ public class ResourceComputationStage extends AbstractBaseStage {
if (currentState.getStateModelDefRef() == null) {
LOG.error("state model def is null." + "resource:" + currentState.getResourceName()
- + ", partitions: " + currentState.getPartitionStateStringMap().keySet() + ", states: "
- + currentState.getPartitionStateStringMap().values());
+ + ", partitions: " + currentState.getPartitionStateStringMap().keySet()
+ + ", states: " + currentState.getPartitionStateStringMap().values());
throw new StageException("State model def is null for resource:"
+ currentState.getResourceName());
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
index 85ae163..b0b6cee 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/TaskAssignmentStage.java
@@ -30,6 +30,7 @@ import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.api.ResourceId;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.LiveInstance;
@@ -89,14 +90,14 @@ public class TaskAssignmentStage extends AbstractBaseStage {
Iterator<Message> iter = messages.iterator();
while (iter.hasNext()) {
Message message = iter.next();
- String resourceName = message.getResourceName();
- Resource resource = resourceMap.get(resourceName);
+ ResourceId resourceId = message.getResourceId();
+ Resource resource = resourceMap.get(resourceId.stringify());
String instanceName = message.getTgtName();
LiveInstance liveInstance = liveInstanceMap.get(instanceName);
String participantVersion = null;
if (liveInstance != null) {
- participantVersion = liveInstance.getHelixVersionString();
+ participantVersion = liveInstance.getHelixVersion().toString();
}
if (resource == null || !resource.getBatchMessageMode() || participantVersion == null
@@ -106,9 +107,9 @@ public class TaskAssignmentStage extends AbstractBaseStage {
}
String key =
- keyBuilder.currentState(message.getTgtName(), message.getTgtSessionIdString(),
- message.getResourceName()).getPath()
- + "/" + message.getFromStateString() + "/" + message.getToStateString();
+ keyBuilder.currentState(message.getTgtName(), message.getTgtSessionId().stringify(),
+ message.getResourceId().stringify()).getPath()
+ + "/" + message.getFromState() + "/" + message.getToState();
if (!batchMessages.containsKey(key)) {
Message batchMessage = new Message(message.getRecord());
@@ -116,7 +117,7 @@ public class TaskAssignmentStage extends AbstractBaseStage {
outputMessages.add(batchMessage);
batchMessages.put(key, batchMessage);
}
- batchMessages.get(key).addPartitionName(message.getPartitionName());
+ batchMessages.get(key).addPartitionName(message.getPartitionId().stringify());
}
return outputMessages;
@@ -131,9 +132,9 @@ public class TaskAssignmentStage extends AbstractBaseStage {
List<PropertyKey> keys = new ArrayList<PropertyKey>();
for (Message message : messages) {
- logger.info("Sending Message " + message.getMsgIdString() + " to " + message.getTgtName()
- + " transit " + message.getPartitionName() + "|" + message.getPartitionNames() + " from:"
- + message.getFromStateString() + " to:" + message.getToStateString());
+ logger.info("Sending Message " + message.getMsgId() + " to " + message.getTgtName()
+ + " transit " + message.getPartitionId() + "|" + message.getPartitionIds() + " from:"
+ + message.getFromState() + " to:" + message.getToState());
// System.out.println("[dbg] Sending Message " + message.getMsgId() + " to " +
// message.getTgtName()
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
index 0fb6709..9f551c1 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
@@ -150,7 +150,7 @@ public class ControllerManager extends AbstractManager {
LiveInstance leader = _dataAccessor.getProperty(_keyBuilder.controllerLeader());
if (leader != null) {
String leaderName = leader.getInstanceName();
- String sessionId = leader.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/CurStateCarryOverUpdater.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
index 3fd3fb8..b45ea98 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
@@ -21,6 +21,9 @@ package org.apache.helix.manager.zk;
import org.I0Itec.zkclient.DataUpdater;
import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Id;
+import org.apache.helix.api.PartitionId;
+import org.apache.helix.api.State;
import org.apache.helix.model.CurrentState;
/**
@@ -52,15 +55,15 @@ class CurStateCarryOverUpdater implements DataUpdater<ZNRecord> {
curState = new CurrentState(_lastCurState.getId());
// copy all simple fields settings and overwrite session-id to current session
curState.getRecord().setSimpleFields(_lastCurState.getRecord().getSimpleFields());
- curState.setSessionId(_curSessionId);
+ curState.setSessionId(Id.session(_curSessionId));
} else {
curState = new CurrentState(currentData);
}
- for (String partitionName : _lastCurState.getPartitionStateStringMap().keySet()) {
+ for (PartitionId partitionId : _lastCurState.getPartitionStateMap().keySet()) {
// carry-over only when current-state not exist
- if (curState.getState(partitionName) == null) {
- curState.setState(partitionName, _initState);
+ if (curState.getState(partitionId) == null) {
+ curState.setState(partitionId, State.from(_initState));
}
}
return curState.getRecord();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/3be06fa6/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
index 523adba..5f6d083 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
@@ -36,7 +36,7 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ message.getMsgType());
}
@@ -63,18 +63,18 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
String type = _message.getMsgType();
HelixTaskResult result = new HelixTaskResult();
if (!type.equals(MessageType.CONTROLLER_MSG.toString())) {
- throw new HelixException("Unexpected msg type for message " + _message.getMsgIdString()
+ throw new HelixException("Unexpected msg type for message " + _message.getMsgId()
+ " type:" + _message.getMsgType());
}
result.getTaskResultMap().put("ControllerResult",
- "msg " + _message.getMsgIdString() + " from " + _message.getMsgSrc() + " processed");
+ "msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
result.setSuccess(true);
return result;
}
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.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/DefaultParticipantErrorMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
index a147ddb..f297252 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
@@ -79,15 +79,15 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
_logger.info("Instance " + _message.getMsgSrc() + " disabled");
} else if (actionOnError == ActionOnError.DISABLE_PARTITION) {
_manager.getClusterManagmentTool().enablePartition(false, _manager.getClusterName(),
- _message.getMsgSrc(), _message.getResourceName(),
- Arrays.asList(_message.getPartitionName()));
- _logger.info("partition " + _message.getPartitionName() + " disabled");
+ _message.getMsgSrc(), _message.getResourceId().stringify(),
+ Arrays.asList(_message.getPartitionId().stringify()));
+ _logger.info("partition " + _message.getPartitionId() + " disabled");
} else if (actionOnError == ActionOnError.DISABLE_RESOURCE) {
// NOT IMPLEMENTED, or we can disable all partitions
// _manager.getClusterManagmentTool().en(_manager.getClusterName(),
// _manager.getInstanceName(),
// _message.getResourceName(), _message.getPartitionName(), false);
- _logger.info("resource " + _message.getResourceName() + " disabled");
+ _logger.info("resource " + _message.getResourceId() + " disabled");
}
} catch (Exception e) {
_logger.error("", e);
@@ -99,7 +99,7 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgIdString(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMsgId(), e);
}
}
@@ -109,7 +109,7 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMsgIdString() + " type:"
+ throw new HelixException("Unexpected msg type for message " + message.getMsgId() + " type:"
+ message.getMsgType());
}