}
MessageOutput output = new MessageOutput();
for (ResourceId resourceId : resourceMap.keySet()) {
ResourceConfig resourceConfig = resourceMap.get(resourceId);
int bucketSize = resourceConfig.getBucketSize();
RebalancerContext rebalancerCtx =
resourceConfig.getRebalancerConfig().getRebalancerContext(RebalancerContext.class);
StateModelDefinition stateModelDef = stateModelDefMap.get(rebalancerCtx.getStateModelDefId());
ResourceAssignment resourceAssignment =
bestPossibleStateOutput.getResourceAssignment(resourceId);
for (PartitionId subUnitId : resourceAssignment.getMappedPartitionIds()) {
Map<ParticipantId, State> instanceStateMap = resourceAssignment.getReplicaMap(subUnitId);
// we should generate message based on the desired-state priority
// so keep generated messages in a temp map keyed by state
// desired-state->list of generated-messages
Map<State, List<Message>> messageMap = new HashMap<State, List<Message>>();
for (ParticipantId participantId : instanceStateMap.keySet()) {
State desiredState = instanceStateMap.get(participantId);
State currentState =
currentStateOutput.getCurrentState(resourceId, subUnitId, participantId);
if (currentState == null) {
currentState = stateModelDef.getTypedInitialState();
}
if (desiredState.equals(currentState)) {
continue;
}
State pendingState =
currentStateOutput.getPendingState(resourceId, subUnitId, participantId);
// TODO fix it
State nextState = stateModelDef.getNextStateForTransition(currentState, desiredState);
if (nextState == null) {
LOG.error("Unable to find a next state for partition: " + subUnitId
+ " from stateModelDefinition" + stateModelDef.getClass() + " from:" + currentState
+ " to:" + desiredState);
continue;
}
if (pendingState != null) {
if (nextState.equals(pendingState)) {
LOG.debug("Message already exists for " + participantId + " to transit " + subUnitId
+ " from " + currentState + " to " + nextState);
} else if (currentState.equals(pendingState)) {
LOG.info("Message hasn't been removed for " + participantId + " to transit"
+ subUnitId + " to " + pendingState + ", desiredState: " + desiredState);
} else {
LOG.info("IdealState changed before state transition completes for " + subUnitId
+ " on " + participantId + ", pendingState: " + pendingState + ", currentState: "
+ currentState + ", nextState: " + nextState);
}
} else {
// TODO check if instance is alive
SessionId sessionId =
cluster.getLiveParticipantMap().get(participantId).getRunningInstance()
.getSessionId();
RebalancerContext rebalancerContext =
resourceConfig.getRebalancerConfig().getRebalancerContext(RebalancerContext.class);
Message message =
createMessage(manager, resourceId, subUnitId, participantId, currentState,
nextState, sessionId, StateModelDefId.from(stateModelDef.getId()),
rebalancerContext.getStateModelFactoryId(), bucketSize);
// TODO refactor get/set timeout/inner-message
if (rebalancerContext != null
&& rebalancerContext.getStateModelDefId().equalsIgnoreCase(
StateModelDefId.SchedulerTaskQueue)) {
if (resourceConfig.getSubUnitMap().size() > 0) {
// TODO refactor it -- we need a way to read in scheduler tasks a priori
Message innerMsg =
resourceConfig.getSchedulerTaskConfig().getInnerMessage(subUnitId);
if (innerMsg != null) {
message.setInnerMessage(innerMsg);
}
}
}
// Set timeout if needed
String stateTransition =
String.format("%s-%s_%s", currentState, nextState,
Message.Attributes.TIMEOUT.name());
SchedulerTaskConfig schedulerTaskConfig = resourceConfig.getSchedulerTaskConfig();
if (schedulerTaskConfig != null) {
int timeout = schedulerTaskConfig.getTimeout(stateTransition, subUnitId);
if (timeout > 0) {
message.setExecutionTimeout(timeout);
}