in helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java [325:485]
public void invoke(NotificationContext changeContext) throws Exception {
Type type = changeContext.getType();
long start = System.currentTimeMillis();
if (logger.isInfoEnabled()) {
logger.info("{} START: CallbackHandler {}, INVOKE {} listener: {} type: {}",
Thread.currentThread().getId(), _uid, _path, _listener, type);
}
synchronized (this) {
if (!_expectTypes.contains(type)) {
logger.warn("Callback handler {} received event in wrong order. Listener: {}, path: {}, "
+ "expected types: {}, but was {}", _uid, _listener, _path, _expectTypes, type);
return;
}
_expectTypes = nextNotificationType.get(type);
if (type == Type.INIT || type == Type.FINALIZE || changeContext.getIsChildChange()) {
subscribeForChanges(changeContext.getType(), _path, _watchChild);
}
}
// This allows the Helix Manager to work with one change at a time
// TODO: Maybe we don't need to sync on _manager for all types of listener. PCould be a
// potential improvement candidate.
synchronized (_manager) {
if (_changeType == IDEAL_STATE) {
IdealStateChangeListener idealStateChangeListener = (IdealStateChangeListener) _listener;
List<IdealState> idealStates = preFetch(_propertyKey);
idealStateChangeListener.onIdealStateChange(idealStates, changeContext);
} else if (_changeType == INSTANCE_CONFIG) {
if (_listener instanceof ConfigChangeListener) {
ConfigChangeListener configChangeListener = (ConfigChangeListener) _listener;
List<InstanceConfig> configs = preFetch(_propertyKey);
configChangeListener.onConfigChange(configs, changeContext);
} else if (_listener instanceof InstanceConfigChangeListener) {
InstanceConfigChangeListener listener = (InstanceConfigChangeListener) _listener;
List<InstanceConfig> configs = Collections.emptyList();
if (_propertyKey.getParams().length > 2 && _preFetchEnabled) {
// If there are more than 2 params, that means the property key is for a specific instance
// and will not have children.
InstanceConfig config = _accessor.getProperty(_propertyKey);
configs = config != null ? Collections.singletonList(config) : Collections.emptyList();
} else {
configs = preFetch(_propertyKey);
}
listener.onInstanceConfigChange(configs, changeContext);
}
} else if (_changeType == RESOURCE_CONFIG) {
ResourceConfigChangeListener listener = (ResourceConfigChangeListener) _listener;
List<ResourceConfig> configs = preFetch(_propertyKey);
listener.onResourceConfigChange(configs, changeContext);
} else if (_changeType == CUSTOMIZED_STATE_CONFIG) {
CustomizedStateConfigChangeListener listener = (CustomizedStateConfigChangeListener) _listener;
CustomizedStateConfig config = null;
if (_preFetchEnabled) {
config = _accessor.getProperty(_propertyKey);
}
listener.onCustomizedStateConfigChange(config, changeContext);
} else if (_changeType == CLUSTER_CONFIG) {
ClusterConfigChangeListener listener = (ClusterConfigChangeListener) _listener;
ClusterConfig config = null;
if (_preFetchEnabled) {
config = _accessor.getProperty(_propertyKey);
}
listener.onClusterConfigChange(config, changeContext);
} else if (_changeType == CONFIG) {
ScopedConfigChangeListener listener = (ScopedConfigChangeListener) _listener;
List<HelixProperty> configs = preFetch(_propertyKey);
listener.onConfigChange(configs, changeContext);
} else if (_changeType == LIVE_INSTANCE) {
LiveInstanceChangeListener liveInstanceChangeListener =
(LiveInstanceChangeListener) _listener;
List<LiveInstance> liveInstances = preFetch(_propertyKey);
liveInstanceChangeListener.onLiveInstanceChange(liveInstances, changeContext);
} else if (_changeType == CURRENT_STATE) {
CurrentStateChangeListener currentStateChangeListener =
(CurrentStateChangeListener) _listener;
String instanceName = PropertyPathConfig.getInstanceNameFromPath(_path);
List<CurrentState> currentStates = preFetch(_propertyKey);
currentStateChangeListener.onStateChange(instanceName, currentStates, changeContext);
} else if (_changeType == TASK_CURRENT_STATE) {
TaskCurrentStateChangeListener taskCurrentStateChangeListener =
(TaskCurrentStateChangeListener) _listener;
String instanceName = PropertyPathConfig.getInstanceNameFromPath(_path);
List<CurrentState> currentStates = preFetch(_propertyKey);
taskCurrentStateChangeListener
.onTaskCurrentStateChange(instanceName, currentStates, changeContext);
} else if (_changeType == CUSTOMIZED_STATE_ROOT) {
CustomizedStateRootChangeListener customizedStateRootChangeListener =
(CustomizedStateRootChangeListener) _listener;
String instanceName = PropertyPathConfig.getInstanceNameFromPath(_path);
List<String> customizedStateTypes = new ArrayList<>();
if (_preFetchEnabled) {
customizedStateTypes =
_accessor.getChildNames(_accessor.keyBuilder().customizedStatesRoot(instanceName));
}
customizedStateRootChangeListener
.onCustomizedStateRootChange(instanceName, customizedStateTypes, changeContext);
} else if (_changeType == CUSTOMIZED_STATE) {
CustomizedStateChangeListener customizedStateChangeListener =
(CustomizedStateChangeListener) _listener;
String instanceName = PropertyPathConfig.getInstanceNameFromPath(_path);
List<CustomizedState> customizedStates = preFetch(_propertyKey);
customizedStateChangeListener.onCustomizedStateChange(instanceName, customizedStates, changeContext);
} else if (_changeType == MESSAGE) {
MessageListener messageListener = (MessageListener) _listener;
String instanceName = PropertyPathConfig.getInstanceNameFromPath(_path);
List<Message> messages = preFetch(_propertyKey);
messageListener.onMessage(instanceName, messages, changeContext);
} else if (_changeType == MESSAGES_CONTROLLER) {
MessageListener messageListener = (MessageListener) _listener;
List<Message> messages = preFetch(_propertyKey);
messageListener.onMessage(_manager.getInstanceName(), messages, changeContext);
} else if (_changeType == EXTERNAL_VIEW || _changeType == TARGET_EXTERNAL_VIEW) {
ExternalViewChangeListener externalViewListener = (ExternalViewChangeListener) _listener;
List<ExternalView> externalViewList = preFetch(_propertyKey);
externalViewListener.onExternalViewChange(externalViewList, changeContext);
} else if (_changeType == CUSTOMIZED_VIEW_ROOT) {
CustomizedViewRootChangeListener customizedViewRootChangeListener =
(CustomizedViewRootChangeListener) _listener;
List<String> customizedViewTypes = new ArrayList<>();
if (_preFetchEnabled) {
customizedViewTypes = _accessor.getChildNames(_accessor.keyBuilder().customizedViews());
}
customizedViewRootChangeListener.onCustomizedViewRootChange(customizedViewTypes,
changeContext);
} else if (_changeType == CUSTOMIZED_VIEW) {
CustomizedViewChangeListener customizedViewListener = (CustomizedViewChangeListener) _listener;
List<CustomizedView> customizedViewListList = preFetch(_propertyKey);
customizedViewListener.onCustomizedViewChange(customizedViewListList, changeContext);
} else if (_changeType == CONTROLLER) {
ControllerChangeListener controllerChangelistener = (ControllerChangeListener) _listener;
controllerChangelistener.onControllerChange(changeContext);
} else {
logger.warn("Callbackhandler {}, Unknown change type: {}", _uid, _changeType);
}
long end = System.currentTimeMillis();
if (logger.isInfoEnabled()) {
logger.info("{} END:INVOKE CallbackHandler {}, {} listener: {} type: {} Took: {}ms",
Thread.currentThread().getId(), _uid, _path, _listener, type, (end - start));
}
if (_monitor != null) {
_monitor.increaseCallbackCounters(end - start);
}
}
}