{
String type = message.getMsgType();
if (!type.equals(MessageType.STATE_TRANSITION.toString()))
{
throw new HelixException("Unexpected msg type for message " + message.getMsgId()
+ " type:" + message.getMsgType());
}
String partitionKey = message.getPartitionName();
String stateModelName = message.getStateModelDef();
String resourceName = message.getResourceName();
String sessionId = message.getTgtSessionId();
int bucketSize = message.getBucketSize();
if (stateModelName == null)
{
logger.error("message does not contain stateModelDef");
return null;
}
String factoryName = message.getStateModelFactoryName();
if (factoryName == null)
{
factoryName = HelixConstants.DEFAULT_STATE_MODEL_FACTORY;
}
StateModelFactory stateModelFactory =
getStateModelFactory(stateModelName, factoryName);
if (stateModelFactory == null)
{
logger.warn("Cannot find stateModelFactory for model:" + stateModelName
+ " using factoryName:" + factoryName + " for resourceGroup:" + resourceName);
return null;
}
// check if the state model definition exists and cache it
if (!_stateModelDefs.containsKey(stateModelName))
{
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
StateModelDefinition stateModelDef =
accessor.getProperty(keyBuilder.stateModelDef(stateModelName));
if (stateModelDef == null)
{
throw new HelixException("stateModelDef for " + stateModelName
+ " does NOT exists");
}
_stateModelDefs.put(stateModelName, stateModelDef);
}