void postHandleMessage() {
HelixTaskResult taskResult =
(HelixTaskResult) _notificationContext.get(MapKey.HELIX_TASK_RESULT.toString());
Exception exception = taskResult.getException();
PartitionId partitionId = _message.getPartitionId();
ResourceId resource = _message.getResourceId();
SessionId sessionId = _message.getTypedTgtSessionId();
String instanceName = _manager.getInstanceName();
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
int bucketSize = _message.getBucketSize();
ZNRecordBucketizer bucketizer = new ZNRecordBucketizer(bucketSize);
// No need to sync on manager, we are cancel executor in expiry session before start executor in
// 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.getTypedTgtSessionId().stringify().equals(_manager.getSessionId())) {
logger.warn("Session id has changed. Skip postExecutionMessage. Old session "
+ _message.getTypedExecutionSessionId() + " , new session : " + _manager.getSessionId());
return;
}
if (taskResult.isSuccess()) {
// String fromState = message.getFromState();
State toState = _message.getTypedToState();
_currentStateDelta.setState(partitionId, toState);
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.
ZNRecordDelta delta =
new ZNRecordDelta(_currentStateDelta.getRecord(), MergeOperation.SUBTRACT);
// Don't subtract simple fields since they contain stateModelDefRef
delta._record.getSimpleFields().clear();
List<ZNRecordDelta> deltaList = new ArrayList<ZNRecordDelta>();
deltaList.add(delta);
_currentStateDelta.setDeltaList(deltaList);
_stateModelFactory.removeStateModel(partitionId.stringify());
} else {
// if the partition is not to be dropped, update _stateModel to the TO_STATE
_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: "
+ partitionId + ", currentState: " + _stateModel.getCurrentState() + ", message: "
+ _message);
_currentStateDelta.setState(partitionId, State.from(_stateModel.getCurrentState()));
} else {
StateTransitionError error =
new StateTransitionError(ErrorType.INTERNAL, ErrorCode.ERROR, exception);
if (exception instanceof InterruptedException) {
if (_isTimeout) {
error = new StateTransitionError(ErrorType.INTERNAL, ErrorCode.TIMEOUT, exception);
} else {
// State transition interrupted but not caused by timeout. Keep the current
// state in this case
logger
.error("State transition interrupted but not timeout. Not updating state. Partition : "
+ _message.getPartitionId() + " MsgId : " + _message.getMessageId());
return;
}
}
_stateModel.rollbackOnError(_message, _notificationContext, error);
_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.getTypedFromState().toString().equalsIgnoreCase(HelixDefinedState.ERROR.toString())) {
disablePartition();
}
}
}
try {
// Update the ZK current state of the node
PropertyKey key =
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);
} else {
// sub-message of a batch message
ConcurrentHashMap<String, CurrentStateUpdate> csUpdateMap =
(ConcurrentHashMap<String, CurrentStateUpdate>) _notificationContext
.get(MapKey.CURRENT_STATE_UPDATE.toString());
csUpdateMap.put(partitionId.stringify(), new CurrentStateUpdate(key, _currentStateDelta));
}
} catch (Exception e) {
logger.error("Error when updating current-state ", e);
StateTransitionError error =
new StateTransitionError(ErrorType.FRAMEWORK, ErrorCode.ERROR, e);