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 2014/08/04 22:19:22 UTC
[5/7] [HELIX-376] Remove HelixConnection/HelixManager duplicate code
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index f95f6ee..12a3fe6 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -21,27 +21,22 @@ package org.apache.helix.manager.zk;
import java.net.InetAddress;
import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Timer;
-import java.util.concurrent.TimeUnit;
-
-import org.I0Itec.zkclient.IZkStateListener;
-import org.I0Itec.zkclient.ZkConnection;
-import org.apache.helix.BaseDataAccessor;
+
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
-import org.apache.helix.ConfigChangeListener;
import org.apache.helix.ControllerChangeListener;
import org.apache.helix.CurrentStateChangeListener;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixConstants.ChangeType;
+import org.apache.helix.HelixMultiClusterController;
+import org.apache.helix.HelixConnection;
+import org.apache.helix.HelixController;
import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
-import org.apache.helix.HelixTimerTask;
+import org.apache.helix.HelixParticipant;
+import org.apache.helix.HelixRole;
+import org.apache.helix.HelixService;
import org.apache.helix.IdealStateChangeListener;
import org.apache.helix.InstanceConfigChangeListener;
import org.apache.helix.InstanceType;
@@ -50,26 +45,20 @@ import org.apache.helix.LiveInstanceInfoProvider;
import org.apache.helix.MessageListener;
import org.apache.helix.PreConnectCallback;
import org.apache.helix.PropertyKey;
-import org.apache.helix.PropertyKey.Builder;
-import org.apache.helix.PropertyPathConfig;
-import org.apache.helix.PropertyType;
import org.apache.helix.ScopedConfigChangeListener;
import org.apache.helix.ZNRecord;
-import org.apache.helix.controller.GenericHelixController;
-import org.apache.helix.messaging.DefaultMessagingService;
+import org.apache.helix.api.id.AdministratorId;
+import org.apache.helix.api.id.ClusterId;
+import org.apache.helix.api.id.ControllerId;
+import org.apache.helix.api.id.ParticipantId;
+import org.apache.helix.api.id.SessionId;
+import org.apache.helix.api.id.SpectatorId;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
-import org.apache.helix.model.LiveInstance;
-import org.apache.helix.monitoring.ZKPathDataDumpTask;
-import org.apache.helix.participant.HelixStateMachineEngine;
import org.apache.helix.participant.StateMachineEngine;
-import org.apache.helix.store.zk.AutoFallbackPropertyStore;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.log4j.Logger;
-import org.apache.zookeeper.Watcher.Event.EventType;
-import org.apache.zookeeper.Watcher.Event.KeeperState;
-import org.apache.zookeeper.ZooKeeper.States;
-public class ZKHelixManager implements HelixManager, IZkStateListener {
+public class ZKHelixManager implements HelixManager {
private static Logger LOG = Logger.getLogger(ZKHelixManager.class);
public static final int FLAPPING_TIME_WINDIOW = 300000; // Default to 300 sec
@@ -77,89 +66,11 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
public static final String ALLOW_PARTICIPANT_AUTO_JOIN = "allowParticipantAutoJoin";
protected final String _zkAddress;
- private final String _clusterName;
- private final String _instanceName;
- private final InstanceType _instanceType;
- private final int _sessionTimeout;
- private final List<PreConnectCallback> _preConnectCallbacks;
- protected final List<CallbackHandler> _handlers;
- private final HelixManagerProperties _properties;
-
- /**
- * helix version#
- */
- private final String _version;
-
- protected ZkClient _zkclient = null;
- private final DefaultMessagingService _messagingService;
-
- private BaseDataAccessor<ZNRecord> _baseDataAccessor;
- private ZKHelixDataAccessor _dataAccessor;
- private final Builder _keyBuilder;
- private ConfigAccessor _configAccessor;
- private ZkHelixPropertyStore<ZNRecord> _helixPropertyStore;
- protected LiveInstanceInfoProvider _liveInstanceInfoProvider = null;
-
- private volatile String _sessionId;
-
- /**
- * Keep track of timestamps that zk State has become Disconnected
- * If in a _timeWindowLengthMs window zk State has become Disconnected
- * for more than_maxDisconnectThreshold times disconnect the zkHelixManager
- */
- private final List<Long> _disconnectTimeHistory = new ArrayList<Long>();
- private final int _flappingTimeWindowMs;
- private final int _maxDisconnectThreshold;
-
- /**
- * participant fields
- */
- private final StateMachineEngine _stateMachineEngine;
- private final List<HelixTimerTask> _timerTasks = new ArrayList<HelixTimerTask>();
-
- /**
- * controller fields
- */
- private final GenericHelixController _controller = new GenericHelixController();
- private CallbackHandler _leaderElectionHandler = null;
- protected final List<HelixTimerTask> _controllerTimerTasks = new ArrayList<HelixTimerTask>();
-
- /**
- * status dump timer-task
- */
- static class StatusDumpTask extends HelixTimerTask {
- Timer _timer = null;
- final HelixManager helixController;
-
- public StatusDumpTask(HelixManager helixController) {
- this.helixController = helixController;
- }
-
- @Override
- public void start() {
- long initialDelay = 0;
- long period = 15 * 60 * 1000;
- long timeThresholdNoChangeForStatusUpdates = 15 * 60 * 1000; // 15 minutes
- long timeThresholdNoChangeForErrors = 24 * 60 * 60 * 1000; // 1 day
- int maximumNumberOfLeafNodesAllowed = 10000;
-
- if (_timer == null) {
- LOG.info("Start StatusDumpTask");
- _timer = new Timer("StatusDumpTimerTask", true);
- _timer.scheduleAtFixedRate(new ZKPathDataDumpTask(helixController,
- timeThresholdNoChangeForStatusUpdates, timeThresholdNoChangeForErrors,
- maximumNumberOfLeafNodesAllowed), initialDelay, period);
- }
- }
+ protected final HelixRole _role; // keep it protected for test purpose
- @Override
- public void stop() {
- if (_timer != null) {
- LOG.info("Stop StatusDumpTask");
- _timer.cancel();
- _timer = null;
- }
- }
+ public ZKHelixManager(HelixRole role) {
+ _role = role;
+ _zkAddress = null;
}
public ZKHelixManager(String clusterName, String instanceName, InstanceType instanceType,
@@ -169,9 +80,7 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
+ clusterName + ", instanceName: " + instanceName + ", type: " + instanceType);
_zkAddress = zkAddress;
- _clusterName = clusterName;
- _instanceType = instanceType;
-
+ ClusterId clusterId = ClusterId.from(clusterName);
if (instanceName == null) {
try {
instanceName =
@@ -183,678 +92,281 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
}
- _instanceName = instanceName;
- _preConnectCallbacks = new ArrayList<PreConnectCallback>();
- _handlers = new ArrayList<CallbackHandler>();
- _properties = new HelixManagerProperties("cluster-manager-version.properties");
- _version = _properties.getVersion();
-
- _keyBuilder = new Builder(clusterName);
- _messagingService = new DefaultMessagingService(this);
-
- /**
- * use system property if available
- */
- _flappingTimeWindowMs =
- getSystemPropertyAsInt("helixmanager.flappingTimeWindow",
- ZKHelixManager.FLAPPING_TIME_WINDIOW);
-
- _maxDisconnectThreshold =
- getSystemPropertyAsInt("helixmanager.maxDisconnectThreshold",
- ZKHelixManager.MAX_DISCONNECT_THRESHOLD);
+ ZkHelixConnection conn = new ZkHelixConnection(zkAddress);
+ conn.connect();
- _sessionTimeout =
- getSystemPropertyAsInt("zk.session.timeout", ZkClient.DEFAULT_SESSION_TIMEOUT);
-
- /**
- * instance type specific init
- */
switch (instanceType) {
- case PARTICIPANT:
- _stateMachineEngine = new HelixStateMachineEngine(this);
+ case PARTICIPANT: {
+ _role = conn.createParticipant(clusterId, ParticipantId.from(instanceName));
break;
- case CONTROLLER:
- _stateMachineEngine = null;
- _controllerTimerTasks.add(new StatusDumpTask(this));
-
+ }
+ case CONTROLLER: {
+ _role = conn.createController(clusterId, ControllerId.from(instanceName));
break;
- case CONTROLLER_PARTICIPANT:
- _stateMachineEngine = new HelixStateMachineEngine(this);
- _controllerTimerTasks.add(new StatusDumpTask(this));
+ }
+ case CONTROLLER_PARTICIPANT: {
+ _role = conn.createAutoController(clusterId, ControllerId.from(instanceName));
break;
- case ADMINISTRATOR:
- case SPECTATOR:
- _stateMachineEngine = null;
+ }
+ case ADMINISTRATOR: {
+ _role = new ZkHelixRoleDefaultImpl(conn, clusterId, AdministratorId.from(instanceName));
break;
- default:
- throw new IllegalArgumentException("unrecognized type: " + instanceType);
}
- }
-
- private int getSystemPropertyAsInt(String propertyKey, int propertyDefaultValue) {
- String valueString = System.getProperty(propertyKey, "" + propertyDefaultValue);
-
- try {
- int value = Integer.parseInt(valueString);
- if (value > 0) {
- return value;
- }
- } catch (NumberFormatException e) {
- LOG.warn("Exception while parsing property: " + propertyKey + ", string: " + valueString
- + ", using default value: " + propertyDefaultValue);
+ case SPECTATOR: {
+ _role = new ZkHelixRoleDefaultImpl(conn, clusterId, SpectatorId.from(instanceName));
+ break;
+ }
+ default:
+ throw new IllegalArgumentException("Unrecognized type: " + instanceType);
}
-
- return propertyDefaultValue;
}
@Override
- public boolean removeListener(PropertyKey key, Object listener) {
- LOG.info("Removing listener: " + listener + " on path: " + key.getPath() + " from cluster: "
- + _clusterName + " by instance: " + _instanceName);
-
- synchronized (this) {
- List<CallbackHandler> toRemove = new ArrayList<CallbackHandler>();
- for (CallbackHandler handler : _handlers) {
- // compare property-key path and listener reference
- if (handler.getPath().equals(key.getPath()) && handler.getListener().equals(listener)) {
- toRemove.add(handler);
- }
- }
-
- _handlers.removeAll(toRemove);
+ public void connect() throws Exception {
+ HelixConnection conn = (ZkHelixConnection) _role.getConnection();
- // handler.reset() may modify the handlers list, so do it outside the iteration
- for (CallbackHandler handler : toRemove) {
- handler.reset();
- }
+ if (!conn.isConnected()) {
+ conn.connect();
}
- return true;
+ HelixService service = (HelixService) _role;
+ service.start();
}
- void checkConnected() {
- if (!isConnected()) {
- throw new HelixException("HelixManager is not connected. Call HelixManager#connect()");
- }
+ @Override
+ public boolean isConnected() {
+ return ((HelixService) _role).isStarted();
}
- void addListener(Object listener, PropertyKey propertyKey, ChangeType changeType,
- EventType[] eventType) {
- checkConnected();
-
- PropertyType type = propertyKey.getType();
-
- synchronized (this) {
- for (CallbackHandler handler : _handlers) {
- // compare property-key path and listener reference
- if (handler.getPath().equals(propertyKey.getPath())
- && handler.getListener().equals(listener)) {
- LOG.info("Listener: " + listener + " on path: " + propertyKey.getPath()
- + " already exists. skip add");
-
- return;
- }
- }
+ @Override
+ public void disconnect() {
+ HelixService service = (HelixService) _role;
+ HelixConnection conn = _role.getConnection();
- CallbackHandler newHandler =
- new CallbackHandler(this, _zkclient, propertyKey, listener, eventType, changeType);
+ service.stop();
- _handlers.add(newHandler);
- LOG.info("Added listener: " + listener + " for type: " + type + " to path: "
- + newHandler.getPath());
+ if (conn.isConnected()) {
+ conn.disconnect();
}
}
@Override
- public void addIdealStateChangeListener(final IdealStateChangeListener listener) throws Exception {
- addListener(listener, new Builder(_clusterName).idealStates(), ChangeType.IDEAL_STATE,
- new EventType[] {
- EventType.NodeDataChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ public void addIdealStateChangeListener(IdealStateChangeListener listener) throws Exception {
+ _role.getConnection().addIdealStateChangeListener(_role, listener, _role.getClusterId());
}
@Override
public void addLiveInstanceChangeListener(LiveInstanceChangeListener listener) throws Exception {
- addListener(listener, new Builder(_clusterName).liveInstances(), ChangeType.LIVE_INSTANCE,
- new EventType[] {
- EventType.NodeDataChanged, EventType.NodeChildrenChanged, EventType.NodeDeleted,
- EventType.NodeCreated
- });
- }
-
- @Override
- public void addConfigChangeListener(ConfigChangeListener listener) throws Exception {
- addListener(listener, new Builder(_clusterName).instanceConfigs(), ChangeType.INSTANCE_CONFIG,
- new EventType[] {
- EventType.NodeChildrenChanged
- });
+ _role.getConnection().addLiveInstanceChangeListener(_role, listener, _role.getClusterId());
}
@Override
public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener)
throws Exception {
- addListener(listener, new Builder(_clusterName).instanceConfigs(), ChangeType.INSTANCE_CONFIG,
- new EventType[] {
- EventType.NodeChildrenChanged
- });
+ _role.getConnection().addInstanceConfigChangeListener(_role, listener, _role.getClusterId());
}
@Override
public void addConfigChangeListener(ScopedConfigChangeListener listener, ConfigScopeProperty scope)
throws Exception {
- Builder keyBuilder = new Builder(_clusterName);
-
- PropertyKey propertyKey = null;
- switch (scope) {
- case CLUSTER:
- propertyKey = keyBuilder.clusterConfigs();
- break;
- case PARTICIPANT:
- propertyKey = keyBuilder.instanceConfigs();
- break;
- case RESOURCE:
- propertyKey = keyBuilder.resourceConfigs();
- break;
- default:
- break;
- }
-
- if (propertyKey != null) {
- addListener(listener, propertyKey, ChangeType.CONFIG, new EventType[] {
- EventType.NodeChildrenChanged
- });
- } else {
- LOG.error("Can't add listener to config scope: " + scope);
- }
- }
-
- // TODO: Decide if do we still need this since we are exposing
- // ClusterMessagingService
- @Override
- public void addMessageListener(MessageListener listener, String instanceName) {
- addListener(listener, new Builder(_clusterName).messages(instanceName), ChangeType.MESSAGE,
- new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ _role.getConnection().addConfigChangeListener(_role, listener, _role.getClusterId(), scope);
}
@Override
- public void addControllerMessageListener(MessageListener listener) {
- addListener(listener, new Builder(_clusterName).controllerMessages(),
- ChangeType.MESSAGES_CONTROLLER, new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ public void addMessageListener(MessageListener listener, String instanceName) throws Exception {
+ _role.getConnection().addMessageListener(_role, listener, _role.getClusterId(),
+ ParticipantId.from(instanceName));
}
@Override
public void addCurrentStateChangeListener(CurrentStateChangeListener listener,
String instanceName, String sessionId) throws Exception {
- addListener(listener, new Builder(_clusterName).currentStates(instanceName, sessionId),
- ChangeType.CURRENT_STATE, new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ _role.getConnection().addCurrentStateChangeListener(_role, listener, _role.getClusterId(),
+ ParticipantId.from(instanceName), SessionId.from(sessionId));
}
@Override
public void addExternalViewChangeListener(ExternalViewChangeListener listener) throws Exception {
- addListener(listener, new Builder(_clusterName).externalViews(), ChangeType.EXTERNAL_VIEW,
- new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ _role.getConnection().addExternalViewChangeListener(_role, listener, _role.getClusterId());
}
@Override
public void addControllerListener(ControllerChangeListener listener) {
- addListener(listener, new Builder(_clusterName).controller(), ChangeType.CONTROLLER,
- new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- });
+ _role.getConnection().addControllerListener(_role, listener, _role.getClusterId());
}
@Override
- public HelixDataAccessor getHelixDataAccessor() {
- checkConnected();
- return _dataAccessor;
+ public void addControllerMessageListener(MessageListener listener) {
+ _role.getConnection().addControllerMessageListener(_role, listener, _role.getClusterId());
}
@Override
- public ConfigAccessor getConfigAccessor() {
- checkConnected();
- return _configAccessor;
+ public boolean removeListener(PropertyKey key, Object listener) {
+ return _role.getConnection().removeListener(_role, listener, key);
}
@Override
- public String getClusterName() {
- return _clusterName;
+ public HelixDataAccessor getHelixDataAccessor() {
+ return _role.getAccessor();
}
@Override
- public String getInstanceName() {
- return _instanceName;
- }
-
- BaseDataAccessor<ZNRecord> createBaseDataAccessor() {
- ZkBaseDataAccessor<ZNRecord> baseDataAccessor = new ZkBaseDataAccessor<ZNRecord>(_zkclient);
-
- return baseDataAccessor;
- }
-
- void createClient() throws Exception {
- PathBasedZkSerializer zkSerializer =
- ChainedPathZkSerializer.builder(new ZNRecordStreamingSerializer()).build();
-
- _zkclient =
- new ZkClient(_zkAddress, _sessionTimeout, ZkClient.DEFAULT_CONNECTION_TIMEOUT, zkSerializer);
-
- _baseDataAccessor = createBaseDataAccessor();
-
- _dataAccessor = new ZKHelixDataAccessor(_clusterName, _instanceType, _baseDataAccessor);
- _configAccessor = new ConfigAccessor(_zkclient);
-
- int retryCount = 0;
-
- _zkclient.subscribeStateChanges(this);
- while (retryCount < 3) {
- try {
- _zkclient.waitUntilConnected(_sessionTimeout, TimeUnit.MILLISECONDS);
- handleStateChanged(KeeperState.SyncConnected);
- handleNewSession();
- break;
- } catch (HelixException e) {
- LOG.error("fail to createClient.", e);
- throw e;
- } catch (Exception e) {
- retryCount++;
-
- LOG.error("fail to createClient. retry " + retryCount, e);
- if (retryCount == 3) {
- throw e;
- }
- }
- }
+ public ConfigAccessor getConfigAccessor() {
+ return _role.getConnection().getConfigAccessor();
}
@Override
- public void connect() throws Exception {
- LOG.info("ClusterManager.connect()");
- if (isConnected()) {
- LOG.warn("Cluster manager: " + _instanceName + " for cluster: " + _clusterName
- + " already connected. skip connect");
- return;
- }
-
- try {
- createClient();
- _messagingService.onConnected();
- } catch (Exception e) {
- LOG.error("fail to connect " + _instanceName, e);
- disconnect();
- throw e;
- }
+ public String getClusterName() {
+ return _role.getClusterId().stringify();
}
@Override
- public void disconnect() {
- if (_zkclient == null) {
- LOG.info("instanceName: " + _instanceName + " already disconnected");
- return;
- }
-
- LOG.info("disconnect " + _instanceName + "(" + _instanceType + ") from " + _clusterName);
-
- try {
- /**
- * stop all timer tasks
- */
- stopTimerTasks();
-
- /**
- * shutdown thread pool first to avoid reset() being invoked in the middle of state
- * transition
- */
- _messagingService.getExecutor().shutdown();
-
- // TODO reset user defined handlers only
- resetHandlers();
-
- if (_leaderElectionHandler != null) {
- _leaderElectionHandler.reset();
- }
-
- } finally {
- _zkclient.close();
- _zkclient = null;
- LOG.info("Cluster manager: " + _instanceName + " disconnected");
- }
+ public String getInstanceName() {
+ return _role.getId().stringify();
}
@Override
public String getSessionId() {
- checkConnected();
- return _sessionId;
- }
-
- @Override
- public boolean isConnected() {
- if (_zkclient == null) {
- return false;
- }
- ZkConnection zkconnection = (ZkConnection) _zkclient.getConnection();
- if (zkconnection != null) {
- States state = zkconnection.getZookeeperState();
- return state == States.CONNECTED;
- }
- return false;
+ return _role.getConnection().getSessionId().stringify();
}
@Override
public long getLastNotificationTime() {
+ // TODO implement this
return 0;
}
@Override
- public void addPreConnectCallback(PreConnectCallback callback) {
- LOG.info("Adding preconnect callback: " + callback);
- _preConnectCallbacks.add(callback);
+ public HelixAdmin getClusterManagmentTool() {
+ return _role.getConnection().createClusterManagementTool();
}
@Override
- public boolean isLeader() {
- if (_instanceType != InstanceType.CONTROLLER
- && _instanceType != InstanceType.CONTROLLER_PARTICIPANT) {
- return false;
- }
-
- if (!isConnected()) {
- return false;
- }
-
- try {
- LiveInstance leader = _dataAccessor.getProperty(_keyBuilder.controllerLeader());
- if (leader != null) {
- String leaderName = leader.getInstanceName();
- String sessionId = leader.getSessionId();
- if (leaderName != null && leaderName.equals(_instanceName) && sessionId != null
- && sessionId.equals(_sessionId)) {
- return true;
- }
- }
- } catch (Exception e) {
- // log
- }
- return false;
- }
-
- @Override
- public synchronized ZkHelixPropertyStore<ZNRecord> getHelixPropertyStore() {
- checkConnected();
-
- if (_helixPropertyStore == null) {
- String path = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, _clusterName);
- String fallbackPath = String.format("/%s/%s", _clusterName, "HELIX_PROPERTYSTORE");
- _helixPropertyStore =
- new AutoFallbackPropertyStore<ZNRecord>(new ZkBaseDataAccessor<ZNRecord>(_zkclient),
- path, fallbackPath);
- }
-
- return _helixPropertyStore;
- }
-
- @Override
- public synchronized HelixAdmin getClusterManagmentTool() {
- checkConnected();
- if (_zkclient != null) {
- return new ZKHelixAdmin(_zkclient);
- }
-
- LOG.error("Couldn't get ZKClusterManagementTool because zkclient is null");
- return null;
+ public ZkHelixPropertyStore<ZNRecord> getHelixPropertyStore() {
+ return (ZkHelixPropertyStore<ZNRecord>) _role.getConnection().createPropertyStore(
+ _role.getClusterId());
}
@Override
public ClusterMessagingService getMessagingService() {
- // The caller can register message handler factories on messaging service before the
- // helix manager is connected. Thus we do not do connected check here.
- return _messagingService;
+ return _role.getMessagingService();
}
@Override
public InstanceType getInstanceType() {
- return _instanceType;
+ return _role.getType();
}
@Override
public String getVersion() {
- return _version;
+ return _role.getConnection().getHelixVersion();
}
@Override
public HelixManagerProperties getProperties() {
- return _properties;
+ return _role.getConnection().getHelixProperties();
}
@Override
public StateMachineEngine getStateMachineEngine() {
- return _stateMachineEngine;
- }
-
- // TODO: rename this and not expose this function as part of interface
- @Override
- public void startTimerTasks() {
- for (HelixTimerTask task : _timerTasks) {
- task.start();
+ StateMachineEngine engine = null;
+ switch (_role.getType()) {
+ case PARTICIPANT: {
+ HelixParticipant participant = (HelixParticipant) _role;
+ engine = participant.getStateMachineEngine();
+ break;
+ }
+ case CONTROLLER_PARTICIPANT: {
+ HelixMultiClusterController autoController = (HelixMultiClusterController) _role;
+ engine = autoController.getStateMachineEngine();
+ break;
+ }
+ default:
+ LOG.info("helix manager type: " + _role.getType() + " does NOT have state-machine-engine");
}
+
+ return engine;
}
@Override
- public void stopTimerTasks() {
- for (HelixTimerTask task : _timerTasks) {
- task.stop();
+ public boolean isLeader() {
+ boolean isLeader = false;
+ switch (_role.getType()) {
+ case CONTROLLER: {
+ HelixController controller = (HelixController) _role;
+ isLeader = controller.isLeader();
+ break;
}
+ case CONTROLLER_PARTICIPANT: {
+ HelixMultiClusterController autoController = (HelixMultiClusterController) _role;
+ isLeader = autoController.isLeader();
+ break;
+ }
+ default:
+ LOG.info("helix manager type: " + _role.getType() + " does NOT support leadership");
+ }
+ return isLeader;
}
@Override
- public void setLiveInstanceInfoProvider(LiveInstanceInfoProvider liveInstanceInfoProvider) {
- _liveInstanceInfoProvider = liveInstanceInfoProvider;
- }
-
- /**
- * wait until we get a non-zero session-id. note that we might lose zkconnection
- * right after we read session-id. but it's ok to get stale session-id and we will have
- * another handle-new-session callback to correct this.
- */
- void waitUntilConnected() {
- boolean isConnected;
- do {
- isConnected =
- _zkclient.waitUntilConnected(ZkClient.DEFAULT_CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
- if (!isConnected) {
- LOG.error("fail to connect zkserver: " + _zkAddress + " in "
- + ZkClient.DEFAULT_CONNECTION_TIMEOUT + "ms. expiredSessionId: " + _sessionId
- + ", clusterName: " + _clusterName);
- continue;
- }
-
- ZkConnection zkConnection = ((ZkConnection) _zkclient.getConnection());
- _sessionId = Long.toHexString(zkConnection.getZookeeper().getSessionId());
-
- /**
- * at the time we read session-id, zkconnection might be lost again
- * wait until we get a non-zero session-id
- */
- } while ("0".equals(_sessionId));
-
- LOG.info("Handling new session, session id: " + _sessionId + ", instance: " + _instanceName
- + ", instanceTye: " + _instanceType + ", cluster: " + _clusterName + ", zkconnection: "
- + ((ZkConnection) _zkclient.getConnection()).getZookeeper());
- }
-
- void initHandlers(List<CallbackHandler> handlers) {
- synchronized (this) {
- if (handlers != null) {
- for (CallbackHandler handler : handlers) {
- handler.init();
- LOG.info("init handler: " + handler.getPath() + ", " + handler.getListener());
- }
- }
+ public void startTimerTasks() {
+ switch (getInstanceType()) {
+ case CONTROLLER: {
+ ((ZkHelixController) _role).startTimerTasks();
+ break;
}
- }
-
- void resetHandlers() {
- synchronized (this) {
- if (_handlers != null) {
- // get a copy of the list and iterate over the copy list
- // in case handler.reset() modify the original handler list
- List<CallbackHandler> tmpHandlers = new ArrayList<CallbackHandler>();
- tmpHandlers.addAll(_handlers);
-
- for (CallbackHandler handler : tmpHandlers) {
- handler.reset();
- LOG.info("reset handler: " + handler.getPath() + ", " + handler.getListener());
- }
- }
+ default:
+ throw new IllegalStateException("Cann't start timer tasks for type: " + getInstanceType());
}
}
- /**
- * If zk state has changed into Disconnected for _maxDisconnectThreshold times during previous
- * _timeWindowLengthMs Ms
- * time window, we think that there are something wrong going on and disconnect the zkHelixManager
- * from zk.
- */
- boolean isFlapping() {
- if (_disconnectTimeHistory.size() == 0) {
- return false;
+ @Override
+ public void stopTimerTasks() {
+ switch (getInstanceType()) {
+ case CONTROLLER: {
+ ((ZkHelixController) _role).stopTimerTasks();
+ break;
}
- long mostRecentTimestamp = _disconnectTimeHistory.get(_disconnectTimeHistory.size() - 1);
-
- // Remove disconnect history timestamp that are older than _flappingTimeWindowMs ago
- while ((_disconnectTimeHistory.get(0) + _flappingTimeWindowMs) < mostRecentTimestamp) {
- _disconnectTimeHistory.remove(0);
+ default:
+ throw new IllegalStateException("Cann't stop timer tasks for type: " + getInstanceType());
}
- return _disconnectTimeHistory.size() > _maxDisconnectThreshold;
}
@Override
- public void handleStateChanged(KeeperState state) throws Exception {
- switch (state) {
- case SyncConnected:
- ZkConnection zkConnection = (ZkConnection) _zkclient.getConnection();
- LOG.info("KeeperState: " + state + ", zookeeper:" + zkConnection.getZookeeper());
- break;
- case Disconnected:
- LOG.info("KeeperState:" + state + ", disconnectedSessionId: " + _sessionId + ", instance: "
- + _instanceName + ", type: " + _instanceType);
-
- /**
- * Track the time stamp that the disconnected happens, then check history and see if
- * we should disconnect the helix-manager
- */
- _disconnectTimeHistory.add(System.currentTimeMillis());
- if (isFlapping()) {
- LOG.error("instanceName: " + _instanceName + " is flapping. diconnect it. "
- + " maxDisconnectThreshold: " + _maxDisconnectThreshold + " disconnects in "
- + _flappingTimeWindowMs + "ms.");
- disconnect();
- }
+ public void addPreConnectCallback(PreConnectCallback callback) {
+ switch (_role.getType()) {
+ case PARTICIPANT: {
+ HelixParticipant participant = (HelixParticipant) _role;
+ participant.addPreConnectCallback(callback);
break;
- case Expired:
- LOG.info("KeeperState:" + state + ", expiredSessionId: " + _sessionId + ", instance: "
- + _instanceName + ", type: " + _instanceType);
+ }
+ case CONTROLLER_PARTICIPANT: {
+ HelixMultiClusterController autoController = (HelixMultiClusterController) _role;
+ autoController.addPreConnectCallback(callback);
break;
+ }
default:
- break;
+ LOG.info("helix manager type: " + _role.getType()
+ + " does NOT support add pre-connect callback");
}
}
@Override
- public void handleNewSession() throws Exception {
- waitUntilConnected();
-
- /**
- * stop all timer tasks, reset all handlers, make sure cleanup completed for previous session
- * disconnect if fail to cleanup
- */
- stopTimerTasks();
- if (_leaderElectionHandler != null) {
- _leaderElectionHandler.reset();
- }
- resetHandlers();
-
- /**
- * clean up write-through cache
- */
- _baseDataAccessor.reset();
-
- /**
- * from here on, we are dealing with new session
- */
- if (!ZKUtil.isClusterSetup(_clusterName, _zkclient)) {
- throw new HelixException("Cluster structure is not set up for cluster: " + _clusterName);
- }
-
- switch (_instanceType) {
- case PARTICIPANT:
- handleNewSessionAsParticipant();
- break;
- case CONTROLLER:
- handleNewSessionAsController();
- break;
- case CONTROLLER_PARTICIPANT:
- handleNewSessionAsParticipant();
- handleNewSessionAsController();
- break;
- case ADMINISTRATOR:
- case SPECTATOR:
- default:
+ public void setLiveInstanceInfoProvider(LiveInstanceInfoProvider liveInstanceInfoProvider) {
+ switch (_role.getType()) {
+ case PARTICIPANT: {
+ HelixParticipant participant = (HelixParticipant) _role;
+ participant.setLiveInstanceInfoProvider(liveInstanceInfoProvider);
break;
}
-
- startTimerTasks();
-
- /**
- * init handlers
- * ok to init message handler and data-accessor twice
- * the second init will be skipped (see CallbackHandler)
- */
- initHandlers(_handlers);
- }
-
- void handleNewSessionAsParticipant() throws Exception {
- /**
- * auto-join
- */
- ParticipantManagerHelper participantHelper =
- new ParticipantManagerHelper(this, _zkclient, _sessionTimeout, _liveInstanceInfoProvider);
- participantHelper.joinCluster();
-
- /**
- * Invoke PreConnectCallbacks
- */
- for (PreConnectCallback callback : _preConnectCallbacks) {
- callback.onPreConnect();
+ case CONTROLLER_PARTICIPANT: {
+ HelixMultiClusterController autoController = (HelixMultiClusterController) _role;
+ autoController.setLiveInstanceInfoProvider(liveInstanceInfoProvider);
+ break;
}
-
- participantHelper.createLiveInstance();
- participantHelper.carryOverPreviousCurrentState();
-
- /**
- * setup message listener
- */
- participantHelper.setupMsgHandler();
- }
-
- void handleNewSessionAsController() {
- if (_leaderElectionHandler != null) {
- _leaderElectionHandler.init();
- } else {
- _leaderElectionHandler =
- new CallbackHandler(this, _zkclient, _keyBuilder.controller(),
- new DistributedLeaderElection(this, _controller, _controllerTimerTasks),
- new EventType[] {
- EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
- }, ChangeType.CONTROLLER);
+ default:
+ LOG.info("helix manager type: " + _role.getType()
+ + " does NOT support set additional live instance information");
}
}
-
}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
index 5961fe3..374a30b 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
@@ -37,7 +37,6 @@ import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.apache.helix.BaseDataAccessor;
-import org.apache.helix.ConfigChangeListener;
import org.apache.helix.ControllerChangeListener;
import org.apache.helix.CurrentStateChangeListener;
import org.apache.helix.ExternalViewChangeListener;
@@ -121,7 +120,7 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
}
_role = role;
- _manager = new HelixConnectionAdaptor(role);
+ _manager = new ZKHelixManager(role);
_instanceName = role.getId().stringify();
_connection = role.getConnection();
_accessor = _connection.createDataAccessor(role.getClusterId());
@@ -171,15 +170,10 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
} else if (_changeType == ChangeType.INSTANCE_CONFIG) {
subscribeForChanges(changeContext, _path, true, true);
- if (_listener instanceof ConfigChangeListener) {
- ConfigChangeListener configChangeListener = (ConfigChangeListener) _listener;
- List<InstanceConfig> configs = _accessor.getChildValues(_propertyKey);
- configChangeListener.onConfigChange(configs, changeContext);
- } else if (_listener instanceof InstanceConfigChangeListener) {
- InstanceConfigChangeListener listener = (InstanceConfigChangeListener) _listener;
- List<InstanceConfig> configs = _accessor.getChildValues(_propertyKey);
- listener.onInstanceConfigChange(configs, changeContext);
- }
+ InstanceConfigChangeListener listener = (InstanceConfigChangeListener) _listener;
+ List<InstanceConfig> configs = _accessor.getChildValues(_propertyKey);
+ listener.onInstanceConfigChange(configs, changeContext);
+
} else if (_changeType == CONFIG) {
subscribeForChanges(changeContext, _path, true, true);
ScopedConfigChangeListener listener = (ScopedConfigChangeListener) _listener;
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixAutoController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixAutoController.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixAutoController.java
deleted file mode 100644
index 1d4b225..0000000
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixAutoController.java
+++ /dev/null
@@ -1,133 +0,0 @@
-package org.apache.helix.manager.zk;
-
-/*
- * 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.ClusterMessagingService;
-import org.apache.helix.HelixAutoController;
-import org.apache.helix.HelixConnection;
-import org.apache.helix.InstanceType;
-import org.apache.helix.LiveInstanceInfoProvider;
-import org.apache.helix.PreConnectCallback;
-import org.apache.helix.api.id.ClusterId;
-import org.apache.helix.api.id.ControllerId;
-import org.apache.helix.api.id.Id;
-import org.apache.helix.api.id.ParticipantId;
-import org.apache.helix.participant.StateMachineEngine;
-import org.apache.log4j.Logger;
-
-public class ZkHelixAutoController implements HelixAutoController {
- private static Logger LOG = Logger.getLogger(ZkHelixAutoController.class);
-
- final ZkHelixConnection _connection;
- final ClusterId _clusterId;
- final ControllerId _controllerId;
- final ZkHelixParticipant _participant;
- final ZkHelixController _controller;
-
- public ZkHelixAutoController(ZkHelixConnection connection, ClusterId clusterId,
- ControllerId controllerId) {
- _connection = connection;
- _clusterId = clusterId;
- _controllerId = controllerId;
-
- _participant =
- new ZkHelixParticipant(connection, clusterId, ParticipantId.from(controllerId.stringify()));
- _controller = new ZkHelixController(connection, clusterId, controllerId);
- }
-
- @Override
- public HelixConnection getConnection() {
- return _connection;
- }
-
- @Override
- public ClusterId getClusterId() {
- return _clusterId;
- }
-
- @Override
- public Id getId() {
- return getControllerId();
- }
-
- @Override
- public InstanceType getType() {
- return InstanceType.CONTROLLER_PARTICIPANT;
- }
-
- @Override
- public ClusterMessagingService getMessagingService() {
- return _participant.getMessagingService();
- }
-
- @Override
- public void start() {
- _connection.addConnectionStateListener(this);
- onConnected();
- }
-
- @Override
- public void stop() {
- _connection.removeConnectionStateListener(this);
- onDisconnecting();
- }
-
- @Override
- public void onConnected() {
- _controller.reset();
- _participant.reset();
-
- _participant.init();
- _controller.init();
- }
-
- @Override
- public void onDisconnecting() {
- LOG.info("disconnecting " + _controllerId + "(" + getType() + ") from " + _clusterId);
- _controller.onDisconnecting();
- _participant.onDisconnecting();
- }
-
- @Override
- public ControllerId getControllerId() {
- return _controllerId;
- }
-
- @Override
- public StateMachineEngine getStateMachineEngine() {
- return _participant.getStateMachineEngine();
- }
-
- @Override
- public void addPreConnectCallback(PreConnectCallback callback) {
- _participant.addPreConnectCallback(callback);
- }
-
- @Override
- public void setLiveInstanceInfoProvider(LiveInstanceInfoProvider liveInstanceInfoProvider) {
- _participant.setLiveInstanceInfoProvider(liveInstanceInfoProvider);
- }
-
- @Override
- public boolean isLeader() {
- return _controller.isLeader();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
index bec6f5c..0a9dc94 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
@@ -27,19 +27,17 @@ import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.helix.BaseDataAccessor;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
-import org.apache.helix.ConfigChangeListener;
import org.apache.helix.ControllerChangeListener;
import org.apache.helix.CurrentStateChangeListener;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixAutoController;
+import org.apache.helix.HelixMultiClusterController;
import org.apache.helix.HelixConnection;
import org.apache.helix.HelixConnectionStateListener;
import org.apache.helix.HelixConstants.ChangeType;
@@ -73,37 +71,35 @@ import org.apache.zookeeper.Watcher.Event.KeeperState;
public class ZkHelixConnection implements HelixConnection, IZkStateListener {
private static Logger LOG = Logger.getLogger(ZkHelixConnection.class);
- final String _zkAddr;
- final int _sessionTimeout;
- SessionId _sessionId;
- ZkClient _zkclient;
- BaseDataAccessor<ZNRecord> _baseAccessor;
- ConfigAccessor _configAccessor;
- final Set<HelixConnectionStateListener> _connectionListener =
- new CopyOnWriteArraySet<HelixConnectionStateListener>();
+ private final String _zkAddr;
+ private final int _sessionTimeout;
+ private SessionId _sessionId;
+ ZkClient _zkclient; // keep it package level for test purpose
+ private BaseDataAccessor<ZNRecord> _baseAccessor;
+ private ConfigAccessor _configAccessor;
+ private final Set<HelixConnectionStateListener> _connectionListener;
- final Map<HelixRole, List<ZkCallbackHandler>> _handlers;
- final HelixManagerProperties _properties;
+ final Map<HelixRole, List<ZkCallbackHandler>> _handlers; // keep it package level for test purpose
+ private final HelixManagerProperties _properties;
/**
* Keep track of timestamps that zk State has become Disconnected
* If in a _timeWindowLengthMs window zk State has become Disconnected
* for more than_maxDisconnectThreshold times disconnect the zkHelixManager
*/
- final List<Long> _disconnectTimeHistory = new ArrayList<Long>();
- final int _flappingTimeWindowMs;
- final int _maxDisconnectThreshold;
-
- final ReentrantLock _lock = new ReentrantLock();
+ private final List<Long> _disconnectTimeHistory = new ArrayList<Long>();
+ private final int _flappingTimeWindowMs;
+ private final int _maxDisconnectThreshold;
/**
* helix version#
*/
- final String _version;
+ private final String _version;
public ZkHelixConnection(String zkAddr) {
_zkAddr = zkAddr;
_handlers = new HashMap<HelixRole, List<ZkCallbackHandler>>();
+ _connectionListener = new CopyOnWriteArraySet<HelixConnectionStateListener>();
/**
* use system property if available
@@ -141,14 +137,16 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
@Override
- public void connect() {
+ public synchronized void connect() {
+ if (isConnected()) {
+ return;
+ }
+
boolean isStarted = false;
try {
- _lock.lock();
_zkclient =
new ZkClient(_zkAddr, ZkClient.DEFAULT_SESSION_TIMEOUT,
ZkClient.DEFAULT_CONNECTION_TIMEOUT, new ZNRecordSerializer());
- // waitUntilConnected();
_baseAccessor = new ZkBaseDataAccessor<ZNRecord>(_zkclient);
_configAccessor = new ConfigAccessor(_zkclient);
@@ -160,7 +158,6 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
} catch (Exception e) {
LOG.error("Exception connect", e);
} finally {
- _lock.unlock();
if (!isStarted) {
disconnect();
}
@@ -168,10 +165,9 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
@Override
- public void disconnect() {
+ public synchronized void disconnect() {
try {
- _lock.lock();
- if (_zkclient == null) {
+ if (!isConnected()) {
return;
}
@@ -195,19 +191,12 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
} catch (Exception e) {
LOG.error("Exception disconnect", e);
- } finally {
- _lock.unlock();
}
}
@Override
- public boolean isConnected() {
- try {
- _lock.lock();
- return _zkclient != null;
- } finally {
- _lock.unlock();
- }
+ public synchronized boolean isConnected() {
+ return _zkclient != null;
}
@Override
@@ -221,8 +210,8 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
@Override
- public HelixAutoController createAutoController(ClusterId clusterId, ControllerId controllerId) {
- return new ZkHelixAutoController(this, clusterId, controllerId);
+ public HelixMultiClusterController createAutoController(ClusterId clusterId, ControllerId controllerId) {
+ return new ZkHelixMultiClusterController(this, clusterId, controllerId);
}
@Override
@@ -243,8 +232,15 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
key.getPath(), fallbackPath);
}
+ private void checkConnected() {
+ if (!isConnected()) {
+ throw new IllegalStateException("Connection not connected. Please call #connect()");
+ }
+ }
+
@Override
public HelixDataAccessor createDataAccessor(ClusterId clusterId) {
+ checkConnected();
return new ZKHelixDataAccessor(clusterId.stringify(), _baseAccessor);
}
@@ -307,16 +303,6 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
@Override
- public void addConfigChangeListener(HelixRole role, ConfigChangeListener listener,
- ClusterId clusterId) {
-
- addListener(role, listener, new PropertyKey.Builder(clusterId.stringify()).instanceConfigs(),
- ChangeType.INSTANCE_CONFIG, new EventType[] {
- EventType.NodeChildrenChanged
- });
- }
-
- @Override
public void addInstanceConfigChangeListener(HelixRole role,
InstanceConfigChangeListener listener, ClusterId clusterId) {
addListener(role, listener, new PropertyKey.Builder(clusterId.stringify()).instanceConfigs(),
@@ -422,57 +408,45 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
}
@Override
- public void handleStateChanged(KeeperState state) throws Exception {
- try {
- _lock.lock();
-
- switch (state) {
- case SyncConnected:
- ZkConnection zkConnection = (ZkConnection) _zkclient.getConnection();
- LOG.info("KeeperState: " + state + ", zookeeper:" + zkConnection.getZookeeper());
- break;
- case Disconnected:
- LOG.info("KeeperState:" + state + ", disconnectedSessionId: " + _sessionId);
-
- /**
- * Track the time stamp that the disconnected happens, then check history and see if
- * we should disconnect the helix-manager
- */
- _disconnectTimeHistory.add(System.currentTimeMillis());
- if (isFlapping()) {
- LOG.error("helix-connection: " + this + ", sessionId: " + _sessionId
- + " is flapping. diconnect it. " + " maxDisconnectThreshold: "
- + _maxDisconnectThreshold + " disconnects in " + _flappingTimeWindowMs + "ms");
- disconnect();
- }
- break;
- case Expired:
- LOG.info("KeeperState:" + state + ", expiredSessionId: " + _sessionId);
- break;
- default:
- break;
+ public synchronized void handleStateChanged(KeeperState state) throws Exception {
+ switch (state) {
+ case SyncConnected:
+ ZkConnection zkConnection = (ZkConnection) _zkclient.getConnection();
+ LOG.info("KeeperState: " + state + ", zookeeper:" + zkConnection.getZookeeper());
+ break;
+ case Disconnected:
+ LOG.info("KeeperState:" + state + ", disconnectedSessionId: " + _sessionId);
+
+ /**
+ * Track the time stamp that the disconnected happens, then check history and see if
+ * we should disconnect the helix-manager
+ */
+ _disconnectTimeHistory.add(System.currentTimeMillis());
+ if (isFlapping()) {
+ LOG.error("helix-connection: " + this + ", sessionId: " + _sessionId
+ + " is flapping. diconnect it. " + " maxDisconnectThreshold: "
+ + _maxDisconnectThreshold + " disconnects in " + _flappingTimeWindowMs + "ms");
+ disconnect();
}
- } finally {
- _lock.unlock();
+ break;
+ case Expired:
+ LOG.info("KeeperState:" + state + ", expiredSessionId: " + _sessionId);
+ break;
+ default:
+ break;
}
}
@Override
- public void handleNewSession() throws Exception {
+ public synchronized void handleNewSession() throws Exception {
waitUntilConnected();
- try {
- _lock.lock();
-
- for (final HelixConnectionStateListener listener : _connectionListener) {
- try {
- listener.onConnected();
- } catch (Exception e) {
- LOG.error("Exception invoking connect on listener: " + listener, e);
- }
+ for (final HelixConnectionStateListener listener : _connectionListener) {
+ try {
+ listener.onConnected();
+ } catch (Exception e) {
+ LOG.error("Exception invoking connect on listener: " + listener, e);
}
- } finally {
- _lock.unlock();
}
}
@@ -514,7 +488,7 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
* at the time we read session-id, zkconnection might be lost again
* wait until we get a non-zero session-id
*/
- } while ("0".equals(_sessionId));
+ } while (!isConnected || "0".equals(_sessionId));
LOG.info("Handling new session, session id: " + _sessionId + ", zkconnection: "
+ ((ZkConnection) _zkclient.getConnection()).getZookeeper());
@@ -527,8 +501,8 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
@Override
public ClusterMessagingService createMessagingService(HelixRole role) {
- HelixManager manager = new HelixConnectionAdaptor(role);
- return new DefaultMessagingService(manager);
+ HelixManager manager = new ZKHelixManager(role);
+ return new DefaultMessagingService(manager);
}
void addListener(HelixRole role, Object listener, PropertyKey propertyKey, ChangeType changeType,
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
index f9529b7..0698945 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
@@ -56,6 +56,7 @@ public class ZkHelixController implements HelixController {
final HelixDataAccessor _accessor;
final HelixManager _manager;
final ZkHelixLeaderElection _leaderElection;
+ boolean _isStarted;
public ZkHelixController(ZkHelixConnection connection, ClusterId clusterId,
ControllerId controllerId) {
@@ -69,7 +70,7 @@ public class ZkHelixController implements HelixController {
_messagingService = (DefaultMessagingService) connection.createMessagingService(this);
_timerTasks = new ArrayList<HelixTimerTask>();
- _manager = new HelixConnectionAdaptor(this);
+ _manager = new ZKHelixManager(this);
_leaderElection = new ZkHelixLeaderElection(this, _pipeline);
_timerTasks.add(new StatusDumpTask(clusterId, _manager.getHelixDataAccessor()));
@@ -104,6 +105,11 @@ public class ZkHelixController implements HelixController {
onDisconnecting();
}
+ @Override
+ public boolean isStarted() {
+ return _isStarted;
+ }
+
void reset() {
/**
* reset all handlers, make sure cleanup completed for previous session
@@ -140,6 +146,7 @@ public class ZkHelixController implements HelixController {
public void onConnected() {
reset();
init();
+ _isStarted = true;
}
@Override
@@ -147,6 +154,8 @@ public class ZkHelixController implements HelixController {
LOG.info("disconnecting " + _controllerId + "(" + getType() + ") from " + _clusterId);
reset();
+
+ _isStarted = false;
}
@Override
@@ -216,7 +225,7 @@ public class ZkHelixController implements HelixController {
/**
* setup generic-controller
*/
- _connection.addConfigChangeListener(this, pipeline, _clusterId);
+ _connection.addInstanceConfigChangeListener(this, pipeline, _clusterId);
_connection.addLiveInstanceChangeListener(this, pipeline, _clusterId);
_connection.addIdealStateChangeListener(this, pipeline, _clusterId);
_connection.addControllerListener(this, pipeline, _clusterId);
@@ -228,7 +237,7 @@ public class ZkHelixController implements HelixController {
}
void removeListenersFromController(GenericHelixController pipeline) {
- PropertyKey.Builder keyBuilder = new PropertyKey.Builder(_manager.getClusterName());
+ PropertyKey.Builder keyBuilder = new PropertyKey.Builder(getClusterId().stringify());
/**
* reset generic-controller
*/
@@ -248,4 +257,9 @@ public class ZkHelixController implements HelixController {
return _manager;
}
+ @Override
+ public HelixDataAccessor getAccessor() {
+ return _accessor;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixMultiClusterController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixMultiClusterController.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixMultiClusterController.java
new file mode 100644
index 0000000..05ad223
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixMultiClusterController.java
@@ -0,0 +1,145 @@
+package org.apache.helix.manager.zk;
+
+/*
+ * 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.ClusterMessagingService;
+import org.apache.helix.HelixMultiClusterController;
+import org.apache.helix.HelixConnection;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.InstanceType;
+import org.apache.helix.LiveInstanceInfoProvider;
+import org.apache.helix.PreConnectCallback;
+import org.apache.helix.api.id.ClusterId;
+import org.apache.helix.api.id.ControllerId;
+import org.apache.helix.api.id.Id;
+import org.apache.helix.api.id.ParticipantId;
+import org.apache.helix.participant.StateMachineEngine;
+import org.apache.log4j.Logger;
+
+public class ZkHelixMultiClusterController implements HelixMultiClusterController {
+ private static Logger LOG = Logger.getLogger(ZkHelixMultiClusterController.class);
+
+ final ZkHelixConnection _connection;
+ final ClusterId _clusterId;
+ final ControllerId _controllerId;
+ final ZkHelixParticipant _participant;
+ final ZkHelixController _controller;
+
+ public ZkHelixMultiClusterController(ZkHelixConnection connection, ClusterId clusterId,
+ ControllerId controllerId) {
+ _connection = connection;
+ _clusterId = clusterId;
+ _controllerId = controllerId;
+
+ _participant =
+ new ZkHelixParticipant(connection, clusterId, ParticipantId.from(controllerId.stringify()));
+ _controller = new ZkHelixController(connection, clusterId, controllerId);
+ }
+
+ @Override
+ public HelixConnection getConnection() {
+ return _connection;
+ }
+
+ @Override
+ public ClusterId getClusterId() {
+ return _clusterId;
+ }
+
+ @Override
+ public Id getId() {
+ return getControllerId();
+ }
+
+ @Override
+ public InstanceType getType() {
+ return InstanceType.CONTROLLER_PARTICIPANT;
+ }
+
+ @Override
+ public ClusterMessagingService getMessagingService() {
+ return _participant.getMessagingService();
+ }
+
+ @Override
+ public void start() {
+ _connection.addConnectionStateListener(this);
+ onConnected();
+ }
+
+ @Override
+ public void stop() {
+ _connection.removeConnectionStateListener(this);
+ onDisconnecting();
+ }
+
+ @Override
+ public boolean isStarted() {
+ return _participant.isStarted() && _controller.isStarted();
+ }
+
+ @Override
+ public void onConnected() {
+ _controller.reset();
+ _participant.reset();
+
+ _participant.init();
+ _controller.init();
+
+ }
+
+ @Override
+ public void onDisconnecting() {
+ LOG.info("disconnecting " + _controllerId + "(" + getType() + ") from " + _clusterId);
+ _controller.onDisconnecting();
+ _participant.onDisconnecting();
+ }
+
+ @Override
+ public ControllerId getControllerId() {
+ return _controllerId;
+ }
+
+ @Override
+ public StateMachineEngine getStateMachineEngine() {
+ return _participant.getStateMachineEngine();
+ }
+
+ @Override
+ public void addPreConnectCallback(PreConnectCallback callback) {
+ _participant.addPreConnectCallback(callback);
+ }
+
+ @Override
+ public void setLiveInstanceInfoProvider(LiveInstanceInfoProvider liveInstanceInfoProvider) {
+ _participant.setLiveInstanceInfoProvider(liveInstanceInfoProvider);
+ }
+
+ @Override
+ public boolean isLeader() {
+ return _controller.isLeader();
+ }
+
+ @Override
+ public HelixDataAccessor getAccessor() {
+ return _participant.getAccessor();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
index d3ee8d1..af50eb7 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
@@ -72,6 +72,7 @@ public class ZkHelixParticipant implements HelixParticipant {
final DefaultMessagingService _messagingService;
final List<PreConnectCallback> _preConnectCallbacks;
final List<HelixTimerTask> _timerTasks;
+ boolean _isStarted;
/**
* state-transition message handler factory for helix-participant
@@ -93,7 +94,7 @@ public class ZkHelixParticipant implements HelixParticipant {
_participantId = participantId;
_messagingService = (DefaultMessagingService) connection.createMessagingService(this);
- HelixManager manager = new HelixConnectionAdaptor(this);
+ HelixManager manager = new ZKHelixManager(this);
_stateMachineEngine = new HelixStateMachineEngine(manager);
_preConnectCallbacks = new ArrayList<PreConnectCallback>();
_timerTasks = new ArrayList<HelixTimerTask>();
@@ -397,6 +398,7 @@ public class ZkHelixParticipant implements HelixParticipant {
public void onConnected() {
reset();
init();
+ _isStarted = true;
}
@Override
@@ -415,12 +417,15 @@ public class ZkHelixParticipant implements HelixParticipant {
* remove live instance ephemeral znode
*/
_accessor.removeProperty(_keyBuilder.liveInstance(_participantId.stringify()));
+ _isStarted = false;
}
@Override
public void start() {
_connection.addConnectionStateListener(this);
- onConnected();
+ if (_connection.isConnected()) {
+ onConnected();
+ }
}
@Override
@@ -430,6 +435,11 @@ public class ZkHelixParticipant implements HelixParticipant {
}
@Override
+ public boolean isStarted() {
+ return _isStarted;
+ }
+
+ @Override
public ClusterMessagingService getMessagingService() {
return _messagingService;
}
@@ -460,6 +470,7 @@ public class ZkHelixParticipant implements HelixParticipant {
_liveInstanceInfoProvider = liveInstanceInfoProvider;
}
+ @Override
public HelixDataAccessor getAccessor() {
return _accessor;
}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixRoleDefaultImpl.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixRoleDefaultImpl.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixRoleDefaultImpl.java
new file mode 100644
index 0000000..9866c7b
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixRoleDefaultImpl.java
@@ -0,0 +1,126 @@
+package org.apache.helix.manager.zk;
+
+/*
+ * 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.BaseDataAccessor;
+import org.apache.helix.ClusterMessagingService;
+import org.apache.helix.ConfigAccessor;
+import org.apache.helix.HelixConnection;
+import org.apache.helix.HelixConnectionStateListener;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixRole;
+import org.apache.helix.HelixService;
+import org.apache.helix.InstanceType;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.api.accessor.ClusterAccessor;
+import org.apache.helix.api.id.ClusterId;
+import org.apache.helix.api.id.Id;
+
+public class ZkHelixRoleDefaultImpl implements HelixRole, HelixService, HelixConnectionStateListener {
+ final ZkHelixConnection _connection;
+ final HelixDataAccessor _accessor;
+ final BaseDataAccessor<ZNRecord> _baseAccessor;
+ final PropertyKey.Builder _keyBuilder;
+ final ClusterAccessor _clusterAccessor;
+ final ConfigAccessor _configAccessor;
+ final ClusterId _clusterId;
+ final Id _instanceId;
+ final ClusterMessagingService _messagingService;
+ boolean _isStarted;
+
+ public ZkHelixRoleDefaultImpl(ZkHelixConnection connection, ClusterId clusterId,
+ Id instanceId) {
+ _connection = connection;
+ _accessor = connection.createDataAccessor(clusterId);
+ _baseAccessor = _accessor.getBaseDataAccessor();
+ _keyBuilder = _accessor.keyBuilder();
+ _clusterAccessor = connection.createClusterAccessor(clusterId);
+ _configAccessor = connection.getConfigAccessor();
+
+ _clusterId = clusterId;
+ _instanceId = instanceId;
+ _messagingService = connection.createMessagingService(this);
+
+ }
+
+ @Override
+ public HelixConnection getConnection() {
+ return _connection;
+ }
+
+ @Override
+ public ClusterId getClusterId() {
+ return _clusterId;
+ }
+
+ @Override
+ public Id getId() {
+ return _instanceId;
+ }
+
+ @Override
+ public InstanceType getType() {
+ return InstanceType.SPECTATOR;
+ }
+
+ @Override
+ public ClusterMessagingService getMessagingService() {
+ return _messagingService;
+ }
+
+ @Override
+ public HelixDataAccessor getAccessor() {
+ return _accessor;
+ }
+
+ @Override
+ public void start() {
+ _connection.addConnectionStateListener(this);
+ if (_connection.isConnected()) {
+ onConnected();
+ }
+ }
+
+ @Override
+ public void stop() {
+ _connection.removeConnectionStateListener(this);
+ onDisconnecting();
+ }
+
+ @Override
+ public boolean isStarted() {
+ return _isStarted;
+ }
+
+ @Override
+ public void onConnected() {
+ _connection.resetHandlers(this);
+ _connection.initHandlers(this);
+ _isStarted = true;
+ }
+
+ @Override
+ public void onDisconnecting() {
+ _connection.resetHandlers(this);
+ _isStarted = false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java b/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
index e799e38..38aaf98 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/DefaultMessagingService.java
@@ -40,11 +40,12 @@ import org.apache.helix.api.id.SessionId;
import org.apache.helix.messaging.handling.AsyncCallbackService;
import org.apache.helix.messaging.handling.HelixTaskExecutor;
import org.apache.helix.messaging.handling.MessageHandlerFactory;
-import org.apache.helix.model.ConfigScope;
+import org.apache.helix.model.HelixConfigScope;
+import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageType;
-import org.apache.helix.model.builder.ConfigScopeBuilder;
+import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.log4j.Logger;
public class DefaultMessagingService implements ClusterMessagingService {
@@ -223,7 +224,7 @@ public class DefaultMessagingService implements ClusterMessagingService {
ConfigAccessor configAccessor = _manager.getConfigAccessor();
if (configAccessor != null) {
- ConfigScope scope = null;
+ HelixConfigScope scope = null;
// Read the participant config and cluster config for the per-message type thread pool size.
// participant config will override the cluster config.
@@ -231,13 +232,16 @@ public class DefaultMessagingService implements ClusterMessagingService {
if (_manager.getInstanceType() == InstanceType.PARTICIPANT
|| _manager.getInstanceType() == InstanceType.CONTROLLER_PARTICIPANT) {
scope =
- new ConfigScopeBuilder().forCluster(_manager.getClusterName())
- .forParticipant(_manager.getInstanceName()).build();
+ new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT)
+ .forCluster(_manager.getClusterName()).forParticipant(_manager.getInstanceName())
+ .build();
threadpoolSizeStr = configAccessor.get(scope, key);
}
if (threadpoolSizeStr == null) {
- scope = new ConfigScopeBuilder().forCluster(_manager.getClusterName()).build();
+ scope =
+ new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(
+ _manager.getClusterName()).build();
threadpoolSizeStr = configAccessor.get(scope, key);
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
index 0ac3b58..8a380c4 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
@@ -493,7 +493,6 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
@Override
public void onMessage(String instanceName, List<Message> messages,
NotificationContext changeContext) {
-
HelixManager manager = changeContext.getManager();
if (_messageQueueMonitor == null) {
_messageQueueMonitor =
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/model/ConfigScope.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/ConfigScope.java b/helix-core/src/main/java/org/apache/helix/model/ConfigScope.java
deleted file mode 100644
index 292ef0f..0000000
--- a/helix-core/src/main/java/org/apache/helix/model/ConfigScope.java
+++ /dev/null
@@ -1,137 +0,0 @@
-package org.apache.helix.model;
-
-/*
- * 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.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
-import org.apache.helix.model.builder.ConfigScopeBuilder;
-import org.apache.helix.util.StringTemplate;
-import org.apache.log4j.Logger;
-
-/**
- * @deprecated replaced by {@link HelixConfigScope}
- */
-@Deprecated
-public class ConfigScope {
- private static Logger LOG = Logger.getLogger(ConfigScope.class);
-
- private static final List<ConfigScopeProperty> scopePriority =
- new ArrayList<ConfigScopeProperty>();
- private static final Map<ConfigScopeProperty, Map<ConfigScopeProperty, ConfigScopeProperty>> scopeTransition =
- new HashMap<ConfigScopeProperty, Map<ConfigScopeProperty, ConfigScopeProperty>>();
- private static final StringTemplate template = new StringTemplate();
- static {
- // scope priority: CLUSTER > PARTICIPANT > RESOURCE > PARTITION
- scopePriority.add(ConfigScopeProperty.CLUSTER);
- scopePriority.add(ConfigScopeProperty.PARTICIPANT);
- scopePriority.add(ConfigScopeProperty.RESOURCE);
- scopePriority.add(ConfigScopeProperty.PARTITION);
-
- // scope transition table to check valid inputs
- scopeTransition.put(ConfigScopeProperty.CLUSTER,
- new HashMap<ConfigScopeProperty, ConfigScopeProperty>());
- scopeTransition.get(ConfigScopeProperty.CLUSTER).put(ConfigScopeProperty.PARTICIPANT,
- ConfigScopeProperty.PARTICIPANT);
- scopeTransition.get(ConfigScopeProperty.CLUSTER).put(ConfigScopeProperty.RESOURCE,
- ConfigScopeProperty.RESOURCE);
- scopeTransition.put(ConfigScopeProperty.RESOURCE,
- new HashMap<ConfigScopeProperty, ConfigScopeProperty>());
- scopeTransition.get(ConfigScopeProperty.RESOURCE).put(ConfigScopeProperty.PARTITION,
- ConfigScopeProperty.PARTITION);
-
- // string templates to generate znode path/index
- // @formatter:off
- template.addEntry(ConfigScopeProperty.CLUSTER, 2,
- "/{clusterName}/CONFIGS/CLUSTER/{clusterName}");
- template.addEntry(ConfigScopeProperty.PARTICIPANT, 2,
- "/{clusterName}/CONFIGS/PARTICIPANT/{participantName}");
- template.addEntry(ConfigScopeProperty.RESOURCE, 2,
- "/{clusterName}/CONFIGS/RESOURCE/{resourceName}");
- template.addEntry(ConfigScopeProperty.PARTITION, 3,
- "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|{partitionName}");
- // @formatter:on
- }
-
- private final String _clusterName;
- private final ConfigScopeProperty _scope;
-
- /**
- * _scopeStr is like:
- * "/ClusterName/CONFIGS/{CLUSTER|PARTICIPANT|RESOURCE|PARTITION}/{clusterName|instanceName|resourceName}{|partitionName}"
- */
- private final String _scopeStr;
-
- public ConfigScope(ConfigScopeBuilder configScopeBuilder) {
- Map<ConfigScopeProperty, String> scopeMap = configScopeBuilder.getScopeMap();
- List<String> keys = new ArrayList<String>();
-
- ConfigScopeProperty curScope = null;
- for (ConfigScopeProperty scope : scopePriority) {
- if (scopeMap.containsKey(scope)) {
- if (curScope == null && scope == ConfigScopeProperty.CLUSTER) {
- keys.add(scopeMap.get(scope));
- curScope = ConfigScopeProperty.CLUSTER;
- } else if (curScope == null) {
- throw new IllegalArgumentException("Missing CLUSTER scope. Can't build scope using "
- + configScopeBuilder);
- } else {
- if (!scopeTransition.containsKey(curScope)
- || !scopeTransition.get(curScope).containsKey(scope)) {
- throw new IllegalArgumentException("Can't build scope using " + configScopeBuilder);
- }
- keys.add(scopeMap.get(scope));
- curScope = scopeTransition.get(curScope).get(scope);
- }
- }
- }
-
- if (curScope == ConfigScopeProperty.CLUSTER) {
- // append one more {clusterName}
- keys.add(scopeMap.get(ConfigScopeProperty.CLUSTER));
- }
-
- String scopeStr = template.instantiate(curScope, keys.toArray(new String[0]));
-
- _clusterName = keys.get(0);
- _scopeStr = scopeStr;
- _scope = curScope;
- }
-
- public ConfigScopeProperty getScope() {
- return _scope;
- }
-
- public String getClusterName() {
- return _clusterName;
- }
-
- public String getScopeStr() {
- return _scopeStr;
- }
-
- @Override
- public String toString() {
- return super.toString() + ": " + _scopeStr;
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/model/builder/ConfigScopeBuilder.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/builder/ConfigScopeBuilder.java b/helix-core/src/main/java/org/apache/helix/model/builder/ConfigScopeBuilder.java
deleted file mode 100644
index a8ce835..0000000
--- a/helix-core/src/main/java/org/apache/helix/model/builder/ConfigScopeBuilder.java
+++ /dev/null
@@ -1,127 +0,0 @@
-package org.apache.helix.model.builder;
-
-/*
- * 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 org.apache.helix.model.ConfigScope;
-import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
-import org.apache.helix.util.StringTemplate;
-import org.apache.log4j.Logger;
-
-/**
- * @deprecated replaced by {@link HelixConfigScopeBuilder}
- */
-@Deprecated
-public class ConfigScopeBuilder {
- private static Logger LOG = Logger.getLogger(ConfigScopeBuilder.class);
-
- private static StringTemplate template = new StringTemplate();
- static {
- // @formatter:off
- template.addEntry(ConfigScopeProperty.CLUSTER, 1, "CLUSTER={clusterName}");
- template.addEntry(ConfigScopeProperty.RESOURCE, 2,
- "CLUSTER={clusterName},RESOURCE={resourceName}");
- template.addEntry(ConfigScopeProperty.PARTITION, 3,
- "CLUSTER={clusterName},RESOURCE={resourceName},PARTITION={partitionName}");
- template.addEntry(ConfigScopeProperty.PARTICIPANT, 2,
- "CLUSTER={clusterName},PARTICIPANT={participantName}");
- // @formatter:on
- }
-
- private final Map<ConfigScopeProperty, String> _scopeMap;
-
- public Map<ConfigScopeProperty, String> getScopeMap() {
- return _scopeMap;
- }
-
- public ConfigScopeBuilder() {
- _scopeMap = new HashMap<ConfigScopeProperty, String>();
- }
-
- public ConfigScopeBuilder forCluster(String clusterName) {
- _scopeMap.put(ConfigScopeProperty.CLUSTER, clusterName);
- return this;
- }
-
- public ConfigScopeBuilder forParticipant(String participantName) {
- _scopeMap.put(ConfigScopeProperty.PARTICIPANT, participantName);
- return this;
- }
-
- public ConfigScopeBuilder forResource(String resourceName) {
- _scopeMap.put(ConfigScopeProperty.RESOURCE, resourceName);
- return this;
-
- }
-
- public ConfigScopeBuilder forPartition(String partitionName) {
- _scopeMap.put(ConfigScopeProperty.PARTITION, partitionName);
- return this;
-
- }
-
- public ConfigScope build() {
- // TODO: validate the scopes map
- return new ConfigScope(this);
- }
-
- public ConfigScope build(ConfigScopeProperty scope, String clusterName, String... scopeKeys) {
- if (scopeKeys == null) {
- scopeKeys = new String[] {};
- }
-
- String[] args = new String[1 + scopeKeys.length];
- args[0] = clusterName;
- System.arraycopy(scopeKeys, 0, args, 1, scopeKeys.length);
- String scopePairs = template.instantiate(scope, args);
-
- return build(scopePairs);
- }
-
- public ConfigScope build(String scopePairs) {
- String[] scopes = scopePairs.split("[\\s,]+");
- for (String scope : scopes) {
- try {
- int idx = scope.indexOf('=');
- if (idx == -1) {
- LOG.error("Invalid scope string: " + scope);
- continue;
- }
-
- String scopeStr = scope.substring(0, idx);
- String value = scope.substring(idx + 1);
- ConfigScopeProperty scopeProperty = ConfigScopeProperty.valueOf(scopeStr);
- _scopeMap.put(scopeProperty, value);
- } catch (Exception e) {
- LOG.error("Invalid scope string: " + scope);
- continue;
- }
- }
-
- return build();
- }
-
- @Override
- public String toString() {
- return _scopeMap.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/participant/CustomCodeInvoker.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/CustomCodeInvoker.java b/helix-core/src/main/java/org/apache/helix/participant/CustomCodeInvoker.java
index a736d71..6c96629 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/CustomCodeInvoker.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/CustomCodeInvoker.java
@@ -21,10 +21,10 @@ package org.apache.helix.participant;
import java.util.List;
-import org.apache.helix.ConfigChangeListener;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceConfigChangeListener;
import org.apache.helix.LiveInstanceChangeListener;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.Type;
@@ -35,7 +35,7 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.log4j.Logger;
-public class CustomCodeInvoker implements LiveInstanceChangeListener, ConfigChangeListener,
+public class CustomCodeInvoker implements LiveInstanceChangeListener, InstanceConfigChangeListener,
ExternalViewChangeListener {
private static Logger LOG = Logger.getLogger(CustomCodeInvoker.class);
private final CustomCodeCallbackHandler _callback;
@@ -89,7 +89,7 @@ public class CustomCodeInvoker implements LiveInstanceChangeListener, ConfigChan
}
@Override
- public void onConfigChange(List<InstanceConfig> configs, NotificationContext changeContext) {
+ public void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext changeContext) {
LOG.info("onConfigChange() invoked");
callParticipantCode(changeContext);
}
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/participant/GenericLeaderStandbyModel.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/GenericLeaderStandbyModel.java b/helix-core/src/main/java/org/apache/helix/participant/GenericLeaderStandbyModel.java
index aa21ee3..3866cf5 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/GenericLeaderStandbyModel.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/GenericLeaderStandbyModel.java
@@ -63,7 +63,7 @@ public class GenericLeaderStandbyModel extends StateModel {
if (notificationType == ChangeType.LIVE_INSTANCE) {
manager.addLiveInstanceChangeListener(_particHolder);
} else if (notificationType == ChangeType.CONFIG) {
- manager.addConfigChangeListener(_particHolder);
+ manager.addInstanceConfigChangeListener(_particHolder);
} else if (notificationType == ChangeType.EXTERNAL_VIEW) {
manager.addExternalViewChangeListener(_particHolder);
} else {
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/main/java/org/apache/helix/spectator/RoutingTableProvider.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/spectator/RoutingTableProvider.java b/helix-core/src/main/java/org/apache/helix/spectator/RoutingTableProvider.java
index 9bba660..7799ca1 100644
--- a/helix-core/src/main/java/org/apache/helix/spectator/RoutingTableProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/spectator/RoutingTableProvider.java
@@ -29,16 +29,16 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.helix.ConfigChangeListener;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.InstanceConfigChangeListener;
import org.apache.helix.NotificationContext;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.InstanceConfig;
import org.apache.log4j.Logger;
-public class RoutingTableProvider implements ExternalViewChangeListener, ConfigChangeListener {
+public class RoutingTableProvider implements ExternalViewChangeListener, InstanceConfigChangeListener {
private static final Logger logger = Logger.getLogger(RoutingTableProvider.class);
private final AtomicReference<RoutingTable> _routingTableRef;
@@ -105,7 +105,7 @@ public class RoutingTableProvider implements ExternalViewChangeListener, ConfigC
}
@Override
- public void onConfigChange(List<InstanceConfig> configs, NotificationContext changeContext) {
+ public void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext changeContext) {
// session has expired clean up the routing table
if (changeContext.getType() == NotificationContext.Type.FINALIZE) {
logger.info("Resetting the routing table. ");
http://git-wip-us.apache.org/repos/asf/helix/blob/02165c52/helix-core/src/test/java/org/apache/helix/Mocks.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/Mocks.java b/helix-core/src/test/java/org/apache/helix/Mocks.java
index a3a6044..0303f12 100644
--- a/helix-core/src/test/java/org/apache/helix/Mocks.java
+++ b/helix-core/src/test/java/org/apache/helix/Mocks.java
@@ -275,12 +275,6 @@ public class Mocks {
}
@Override
- public void addConfigChangeListener(ConfigChangeListener listener) {
- // TODO Auto-generated method stub
-
- }
-
- @Override
public void addMessageListener(MessageListener listener, String instanceName) {
// TODO Auto-generated method stub