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 2013/08/06 22:35:13 UTC
[2/2] git commit: HELIX-195: fix race condition between FINALIZE
callbacks and zk callbacks, rb=13345
HELIX-195: fix race condition between FINALIZE callbacks and zk callbacks, rb=13345
Project: http://git-wip-us.apache.org/repos/asf/incubator-helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-helix/commit/6d539799
Tree: http://git-wip-us.apache.org/repos/asf/incubator-helix/tree/6d539799
Diff: http://git-wip-us.apache.org/repos/asf/incubator-helix/diff/6d539799
Branch: refs/heads/master
Commit: 6d5397990d9629009e304ae6eea4a046d5429871
Parents: 38449e0
Author: zzhang <zz...@apache.org>
Authored: Tue Aug 6 13:34:59 2013 -0700
Committer: zzhang <zz...@apache.org>
Committed: Tue Aug 6 13:34:59 2013 -0700
----------------------------------------------------------------------
.../controller/GenericHelixController.java | 61 ++++---
.../helix/manager/zk/AbstractManager.java | 139 +++++++--------
.../helix/manager/zk/CallbackHandler.java | 77 +++++---
.../helix/manager/zk/ControllerManager.java | 15 +-
.../zk/DistributedControllerManager.java | 15 +-
.../manager/zk/DistributedLeaderElection.java | 16 +-
.../helix/manager/zk/ParticipantManager.java | 14 +-
.../manager/zk/ParticipantManagerHelper.java | 2 +-
.../test/java/org/apache/helix/TestHelper.java | 76 ++++++--
.../manager/ClusterDistributedController.java | 103 +++++++++++
.../manager/TestConsecutiveZkSessionExpiry.java | 112 +++++++++++-
.../manager/TestControllerManager.java | 174 +++++++++++++++++++
.../TestDistributedControllerManager.java | 117 ++++++++++++-
.../manager/TestParticipantManager.java | 2 +-
.../manager/TestZkCallbackHandlerLeak.java | 94 ++--------
15 files changed, 762 insertions(+), 255 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index a898160..58095df 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -101,9 +101,9 @@ public class GenericHelixController implements
final AtomicReference<Map<String, LiveInstance>> _lastSeenInstances;
final AtomicReference<Map<String, LiveInstance>> _lastSeenSessions;
-
+
ClusterStatusMonitor _clusterStatusMonitor;
-
+
/**
* The _paused flag is checked by function handleEvent(), while if the flag is set
@@ -132,12 +132,12 @@ public class GenericHelixController implements
class RebalanceTask extends TimerTask
{
HelixManager _manager;
-
+
public RebalanceTask(HelixManager manager)
{
_manager = manager;
}
-
+
@Override
public void run()
{
@@ -148,15 +148,15 @@ public class GenericHelixController implements
event.addAttribute("changeContext", changeContext);
List<ZNRecord> dummy = new ArrayList<ZNRecord>();
event.addAttribute("eventData", dummy);
- // Should be able to process
+ // Should be able to process
handleEvent(event);
}
}
-
+
// TODO who should stop this timer
/**
* Starts the rebalancing timer with the specified period. Start the timer if necessary;
- * If the period is smaller than the current period, cancel the current timer and use
+ * If the period is smaller than the current period, cancel the current timer and use
* the new period.
*/
void startRebalancingTimer(int period, HelixManager manager)
@@ -177,9 +177,9 @@ public class GenericHelixController implements
logger.info("Controller already has timer at period " + _timerPeriod);
}
}
-
+
/**
- * Starts the rebalancing timer
+ * Starts the rebalancing timer
*/
void stopRebalancingTimer()
{
@@ -190,7 +190,7 @@ public class GenericHelixController implements
}
_timerPeriod = Integer.MAX_VALUE;
}
-
+
private static PipelineRegistry createDefaultRegistry()
{
logger.info("createDefaultRegistry");
@@ -303,7 +303,7 @@ public class GenericHelixController implements
_clusterStatusMonitor.reset();
_clusterStatusMonitor = null;
}
-
+
stopRebalancingTimer();
logger.info("Get FINALIZE notification, skip the pipeline. Event :" + event.getName());
return;
@@ -314,7 +314,7 @@ public class GenericHelixController implements
{
_clusterStatusMonitor = new ClusterStatusMonitor(manager.getClusterName());
}
-
+
event.addAttribute("clusterStatusMonitor", _clusterStatusMonitor);
}
}
@@ -391,19 +391,19 @@ public class GenericHelixController implements
NotificationContext changeContext)
{
logger.info("START: GenericClusterController.onMessage()");
-
+
ClusterEvent event = new ClusterEvent("messageChange");
event.addAttribute("helixmanager", changeContext.getManager());
event.addAttribute("instanceName", instanceName);
event.addAttribute("changeContext", changeContext);
event.addAttribute("eventData", messages);
handleEvent(event);
-
+
if (_clusterStatusMonitor != null && messages != null)
{
_clusterStatusMonitor.addMessageQueueSize(instanceName, messages.size());
}
-
+
logger.info("END: GenericClusterController.onMessage()");
}
@@ -412,6 +412,7 @@ public class GenericHelixController implements
NotificationContext changeContext)
{
logger.info("START: Generic GenericClusterController.onLiveInstanceChange()");
+
if (liveInstances == null)
{
liveInstances = Collections.emptyList();
@@ -422,6 +423,13 @@ public class GenericHelixController implements
changeContext.getType() == NotificationContext.Type.CALLBACK)
{
checkLiveInstancesObservation(liveInstances, changeContext);
+ } else if (changeContext.getType() == NotificationContext.Type.FINALIZE)
+ {
+ // on finalize, should remove all message/current-state listeners
+ logger.info("remove message/current-state listeners. lastSeenInstances: "
+ + _lastSeenInstances + ", lastSeenSessions: " + _lastSeenSessions);
+ liveInstances = Collections.emptyList();
+ checkLiveInstancesObservation(liveInstances, changeContext);
}
ClusterEvent event = new ClusterEvent("liveInstanceChange");
@@ -431,7 +439,7 @@ public class GenericHelixController implements
handleEvent(event);
logger.info("END: Generic GenericClusterController.onLiveInstanceChange()");
}
-
+
void checkRebalancingTimer(HelixManager manager, List<IdealState> idealStates)
{
if (manager.getConfigAccessor() == null)
@@ -439,7 +447,7 @@ public class GenericHelixController implements
logger.warn(manager.getInstanceName() + " config accessor doesn't exist. should be in file-based mode.");
return;
}
-
+
for(IdealState idealState : idealStates)
{
int period = idealState.getRebalanceTimerPeriod();
@@ -449,7 +457,7 @@ public class GenericHelixController implements
}
}
}
-
+
@Override
public void onIdealStateChange(List<IdealState> idealStates,
NotificationContext changeContext)
@@ -460,12 +468,12 @@ public class GenericHelixController implements
event.addAttribute("changeContext", changeContext);
event.addAttribute("eventData", idealStates);
handleEvent(event);
-
+
if(changeContext.getType() != Type.FINALIZE)
{
checkRebalancingTimer(changeContext.getManager(), idealStates);
}
-
+
logger.info("END: Generic GenericClusterController.onIdealStateChange()");
}
@@ -571,11 +579,11 @@ public class GenericHelixController implements
if (!curSessions.containsKey(session)) {
// remove current-state listener for expired session
String instanceName = lastSessions.get(session).getInstanceName();
- manager.removeListener(keyBuilder.currentStates(instanceName, session), this);
+ manager.removeListener(keyBuilder.currentStates(instanceName, session), this);
}
}
}
-
+
if (lastInstances != null) {
for (String instance : lastInstances.keySet()) {
if (!curInstances.containsKey(instance)) {
@@ -584,15 +592,15 @@ public class GenericHelixController implements
}
}
}
-
+
for (String session : curSessions.keySet()) {
if (lastSessions == null || !lastSessions.containsKey(session)) {
String instanceName = curSessions.get(session).getInstanceName();
try {
// add current-state listeners for new sessions
manager.addCurrentStateChangeListener(this, instanceName, session);
- logger.info("Succeed in addling current state listener for instance: " + instanceName + " with session: " + session);
-
+ logger.info(manager.getInstanceName() + " added current-state listener for instance: "
+ + instanceName + ", session: " + session + ", listener: " + this);
} catch (Exception e) {
logger.error("Fail to add current state listener for instance: "
+ instanceName + " with session: " + session, e);
@@ -605,7 +613,8 @@ public class GenericHelixController implements
try {
// add message listeners for new instances
manager.addMessageListener(this, instance);
- logger.info("Succeed in adding message listener for " + instance);
+ logger.info(manager.getInstanceName() + " added message listener for "
+ + instance + ", listener: " + this);
}
catch (Exception e)
{
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/AbstractManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/AbstractManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/AbstractManager.java
index a68e1de..90f553d 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/AbstractManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/AbstractManager.java
@@ -77,7 +77,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
final List<PreConnectCallback> _preConnectCallbacks;
protected final List<CallbackHandler> _handlers;
final HelixManagerProperties _properties;
-
+
/**
* helix version#
*/
@@ -96,21 +96,21 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
volatile String _sessionId;
-
+
/**
* Keep track of timestamps that zk State has become Disconnected
- * If in a _timeWindowLengthMs window 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 LinkedList<Long>();
-
- final int _flappingTimeWindowMs;
+
+ final int _flappingTimeWindowMs;
final int _maxDisconnectThreshold;
-
+
public AbstractManager(String zkAddress, String clusterName, String instanceName,
InstanceType instanceType) {
-
+
LOG.info("Create a zk-based cluster manager. zkSvr: " + zkAddress + ", clusterName: " + clusterName
+ ", instanceName: " + instanceName + ", type: " + instanceType);
@@ -126,23 +126,23 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
_keyBuilder = new Builder(clusterName);
_messagingService = new DefaultMessagingService(this);
-
+
/**
* use system property if available
*/
- _flappingTimeWindowMs = getSystemPropertyAsInt("helixmanager.flappingTimeWindow",
+ _flappingTimeWindowMs = getSystemPropertyAsInt("helixmanager.flappingTimeWindow",
ZKHelixManager.FLAPPING_TIME_WINDIOW);
-
- _maxDisconnectThreshold = getSystemPropertyAsInt("helixmanager.maxDisconnectThreshold",
+
+ _maxDisconnectThreshold = getSystemPropertyAsInt("helixmanager.maxDisconnectThreshold",
ZKHelixManager.MAX_DISCONNECT_THRESHOLD);
-
+
_sessionTimeout = getSystemPropertyAsInt("zk.session.timeout", ZkClient.DEFAULT_SESSION_TIMEOUT);
}
-
+
private int getSystemPropertyAsInt(String propertyKey, int propertyDefaultValue) {
String valueString = System.getProperty(propertyKey, "" + propertyDefaultValue);
-
+
try
{
int value = Integer.parseInt(valueString);
@@ -155,15 +155,15 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
LOG.warn("Exception while parsing property: " + propertyKey + ", string: " + valueString
+ ", using default value: " + propertyDefaultValue);
}
-
+
return propertyDefaultValue;
}
-
+
/**
* different types of helix manager should impl its own handle new session logic
*/
// public abstract void handleNewSession();
-
+
@Override
public void connect() throws Exception {
LOG.info("ClusterManager.connect()");
@@ -184,7 +184,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
LOG.error("fail to connect " + _instanceName, e);
disconnect();
throw e;
- }
+ }
}
@Override
@@ -204,9 +204,9 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
* specific disconnect logic for each helix-manager type
*/
abstract void doDisconnect();
-
+
/**
- * This function can be called when the connection are in bad state(e.g. flapping),
+ * This function can be called when the connection are in bad state(e.g. flapping),
* in which isConnected() could be false and we want to disconnect from cluster.
*/
@Override
@@ -219,50 +219,50 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
* 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();
-
+
_dataAccessor.shutdown();
-
+
doDisconnect();
-
+
_zkclient.unsubscribeAll();
} finally {
_zkclient.close();
LOG.info("Cluster manager: " + _instanceName + " disconnected");
- }
+ }
}
@Override
public void addIdealStateChangeListener(IdealStateChangeListener listener) throws Exception {
- addListener(listener, new Builder(_clusterName).idealStates(), ChangeType.IDEAL_STATE,
- new EventType[] { EventType.NodeDataChanged, EventType.NodeDeleted, EventType.NodeCreated });
+ addListener(listener, new Builder(_clusterName).idealStates(), ChangeType.IDEAL_STATE,
+ new EventType[] { EventType.NodeDataChanged, EventType.NodeDeleted, EventType.NodeCreated });
}
@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 });
+ 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 });
+ addListener(listener, new Builder(_clusterName).instanceConfigs(), ChangeType.INSTANCE_CONFIG,
+ new EventType[] { EventType.NodeChildrenChanged });
}
@Override
public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener)
throws Exception {
- addListener(listener, new Builder(_clusterName).instanceConfigs(), ChangeType.INSTANCE_CONFIG,
- new EventType[] { EventType.NodeChildrenChanged });
+ addListener(listener, new Builder(_clusterName).instanceConfigs(), ChangeType.INSTANCE_CONFIG,
+ new EventType[] { EventType.NodeChildrenChanged });
}
@Override
@@ -288,38 +288,38 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
if (propertyKey != null)
{
- addListener(listener, propertyKey, ChangeType.CONFIG,
+ addListener(listener, propertyKey, ChangeType.CONFIG,
new EventType[] { EventType.NodeChildrenChanged });
} else
{
LOG.error("Can't add listener to config scope: " + scope);
- }
+ }
}
@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 });
+ addListener(listener, new Builder(_clusterName).messages(instanceName), ChangeType.MESSAGE,
+ new EventType[] { EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated });
}
@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 });
+ new EventType[] { EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated });
}
@Override
public void addHealthStateChangeListener(HealthStateChangeListener listener, String instanceName)
throws Exception {
addListener(listener, new Builder(_clusterName).healthReports(instanceName), ChangeType.HEALTH,
- new EventType[] { EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated });
+ new EventType[] { EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated });
}
@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 });
+ new EventType[] { EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated });
}
@Override
@@ -336,7 +336,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
@Override
public boolean removeListener(PropertyKey key, Object listener) {
- LOG.info("Removing listener: " + listener + " on path: " + key.getPath()
+ LOG.info("Removing listener: " + listener + " on path: " + key.getPath()
+ " from cluster: " + _clusterName + " by instance: " + _instanceName);
synchronized (this)
@@ -350,16 +350,16 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
toRemove.add(handler);
}
}
-
+
_handlers.removeAll(toRemove);
-
+
// handler.reset() may modify the handlers list, so do it outside the iteration
for (CallbackHandler handler : toRemove) {
handler.reset();
}
}
- return true;
+ return true;
}
@Override
@@ -403,7 +403,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
{
return new ZKHelixAdmin(_zkclient);
}
-
+
LOG.error("Couldn't get ZKClusterManagementTool because zkclient is null");
return null;
}
@@ -478,24 +478,24 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
{
task.stop();
}
-
+
}
@Override
public void addPreConnectCallback(PreConnectCallback callback) {
LOG.info("Adding preconnect callback: " + callback);
- _preConnectCallbacks.add(callback);
+ _preConnectCallbacks.add(callback);
}
@Override
public void setLiveInstanceInfoProvider(LiveInstanceInfoProvider liveInstanceInfoProvider) {
- _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.
+ * 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.
*/
protected void waitUntilConnected() {
boolean isConnected;
@@ -508,7 +508,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
+ ", clusterName: " + _clusterName);
continue;
}
-
+
ZkConnection zkConnection = ((ZkConnection) _zkclient.getConnection());
_sessionId = Long.toHexString(zkConnection.getZookeeper().getSessionId());
@@ -517,13 +517,13 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
* 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());
}
-
+
protected void checkConnected()
{
if (!isConnected())
@@ -546,18 +546,19 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
if (handler.getPath().equals(propertyKey.getPath()) && handler.getListener().equals(listener))
{
LOG.info("Listener: " + listener + " on path: " + propertyKey.getPath() + " already exists. skip add");
+
return;
}
}
-
- CallbackHandler newHandler = new CallbackHandler(this, _zkclient, propertyKey,
+
+ CallbackHandler newHandler = new CallbackHandler(this, _zkclient, propertyKey,
listener, eventType, changeType);
-
+
_handlers.add(newHandler);
LOG.info("Added listener: " + listener + " for type: " + type + " to path: " + newHandler.getPath());
}
}
-
+
protected void initHandlers(List<CallbackHandler> handlers)
{
synchronized (this)
@@ -572,7 +573,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
}
}
}
-
+
protected void resetHandlers()
{
synchronized (this)
@@ -582,7 +583,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
// in case handler.reset() modify the original handler list
List<CallbackHandler> tmpHandlers = new ArrayList<CallbackHandler>();
tmpHandlers.addAll(_handlers);
-
+
for (CallbackHandler handler : tmpHandlers)
{
handler.reset();
@@ -594,14 +595,14 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
/**
* different helix-manager may override this to have a cache-enabled based-data-accessor
- *
+ *
* @param baseDataAccessor
* @return
*/
BaseDataAccessor<ZNRecord> createBaseDataAccessor(ZkBaseDataAccessor<ZNRecord> baseDataAccessor) {
return baseDataAccessor;
}
-
+
void createClient() throws Exception {
PathBasedZkSerializer zkSerializer =
ChainedPathZkSerializer.builder(new ZNRecordStreamingSerializer())
@@ -616,7 +617,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
_dataAccessor =
new ZKHelixDataAccessor(_clusterName, _instanceType, _baseDataAccessor);
_configAccessor = new ConfigAccessor(_zkclient);
-
+
int retryCount = 0;
_zkclient.subscribeStateChanges(this);
@@ -669,8 +670,8 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
if (isFlapping())
{
LOG.error("instanceName: " + _instanceName + " is flapping. diconnect it. "
- + " maxDisconnectThreshold: " + _maxDisconnectThreshold
- + " disconnects in " + _flappingTimeWindowMs + "ms.");
+ + " maxDisconnectThreshold: " + _maxDisconnectThreshold
+ + " disconnects in " + _flappingTimeWindowMs + "ms.");
disconnect();
}
break;
@@ -680,9 +681,9 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
break;
default:
break;
- }
+ }
}
-
+
/**
* 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.
@@ -694,7 +695,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
return false;
}
long mostRecentTimestamp = _disconnectTimeHistory.get(_disconnectTimeHistory.size() - 1);
-
+
// Remove disconnect history timestamp that are older than _flappingTimeWindowMs ago
while ((_disconnectTimeHistory.get(0) + _flappingTimeWindowMs) < mostRecentTimestamp)
{
@@ -705,7 +706,7 @@ public abstract class AbstractManager implements HelixManager, IZkStateListener
/**
* controller should override it to return a list of timers that need to start/stop when leadership changes
- *
+ *
* @return
*/
protected List<HelixTimerTask> getControllerHelixTimerTasks() {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
index 533650f..b30cee6 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
@@ -27,7 +27,10 @@ import static org.apache.helix.HelixConstants.ChangeType.LIVE_INSTANCE;
import static org.apache.helix.HelixConstants.ChangeType.MESSAGE;
import static org.apache.helix.HelixConstants.ChangeType.MESSAGES_CONTROLLER;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.I0Itec.zkclient.IZkChildListener;
@@ -69,9 +72,19 @@ import org.apache.zookeeper.Watcher.Event.EventType;
public class CallbackHandler implements IZkChildListener, IZkDataListener
{
-
private static Logger logger = Logger.getLogger(CallbackHandler.class);
+ /**
+ * define the next possible notification types
+ */
+ private static Map<Type, List<Type>> nextNotificationType = new HashMap<Type, List<Type>>();
+ static
+ {
+ nextNotificationType.put(Type.INIT, Arrays.asList(Type.CALLBACK, Type.FINALIZE));
+ nextNotificationType.put(Type.CALLBACK, Arrays.asList(Type.CALLBACK, Type.FINALIZE));
+ nextNotificationType.put(Type.FINALIZE, Arrays.asList(Type.INIT));
+ }
+
private final String _path;
private final Object _listener;
private final EventType[] _eventTypes;
@@ -82,6 +95,12 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
private final HelixManager _manager;
private final PropertyKey _propertyKey;
+ /**
+ * maintain the expected notification types
+ * this is fix for HELIX-195: race condition between FINALIZE callbacks and Zk callbacks
+ */
+ private List<NotificationContext.Type> _expectTypes = nextNotificationType.get(Type.FINALIZE);
+
public CallbackHandler(HelixManager manager,
ZkClient client,
PropertyKey propertyKey,
@@ -92,7 +111,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
if (listener == null) {
throw new HelixException("listener could not be null");
}
-
+
this._manager = manager;
this._accessor = manager.getHelixDataAccessor();
this._zkClient = client;
@@ -120,6 +139,14 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
// This allows the listener to work with one change at a time
synchronized (_manager)
{
+ Type type = changeContext.getType();
+ if (!_expectTypes.contains(type)) {
+ logger.warn("Skip processing callbacks for listener: " + _listener + ", path: " + _path
+ + ", expected types: " + _expectTypes + " but was " + type);
+ return;
+ }
+ _expectTypes = nextNotificationType.get(type);
+
// Builder keyBuilder = _accessor.keyBuilder();
long start = System.currentTimeMillis();
if (logger.isInfoEnabled())
@@ -151,8 +178,8 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
{
InstanceConfigChangeListener listener = (InstanceConfigChangeListener) _listener;
List<InstanceConfig> configs = _accessor.getChildValues(_propertyKey);
- listener.onInstanceConfigChange(configs, changeContext);
- }
+ listener.onInstanceConfigChange(configs, changeContext);
+ }
}
else if (_changeType == CONFIG)
{
@@ -254,19 +281,19 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
NotificationContext.Type type = context.getType();
if (type == NotificationContext.Type.INIT || type == NotificationContext.Type.CALLBACK)
{
- logger.info(_manager.getInstanceName() + " subscribes child-change. path: "
+ logger.info(_manager.getInstanceName() + " subscribes child-change. path: "
+ path + ", listener: " + _listener);
_zkClient.subscribeChildChanges(path, this);
}
else if (type == NotificationContext.Type.FINALIZE)
{
- logger.info(_manager.getInstanceName() + " unsubscribe child-change. path: "
+ logger.info(_manager.getInstanceName() + " unsubscribe child-change. path: "
+ path + ", listener: " + _listener);
-
+
_zkClient.unsubscribeChildChanges(path, this);
}
}
-
+
private void subscribeDataChange(String path, NotificationContext context)
{
NotificationContext.Type type = context.getType();
@@ -275,7 +302,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
{
if (logger.isDebugEnabled())
{
- logger.debug(_manager.getInstanceName() + " subscribe data-change. path: "
+ logger.debug(_manager.getInstanceName() + " subscribe data-change. path: "
+ path + ", listener: " + _listener);
}
_zkClient.subscribeDataChanges(path, this);
@@ -283,13 +310,13 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
}
else if (type == NotificationContext.Type.FINALIZE)
{
- logger.info(_manager.getInstanceName() + " unsubscribe data-change. path: "
+ logger.info(_manager.getInstanceName() + " unsubscribe data-change. path: "
+ path + ", listener: " + _listener);
_zkClient.unsubscribeDataChanges(path, this);
}
}
-
+
// TODO watchParent is always true. consider remove it
private void subscribeForChanges(NotificationContext context,
String path,
@@ -326,16 +353,16 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
// data-change gives a delete-callback which is used to remove watch
subscribeChildChange(childPath, context);
subscribeDataChange(childPath, context);
-
+
// subscribe data-change on bucketized child
List<String> bucketizedChildNames = _zkClient.getChildren(childPath);
- if (bucketizedChildNames != null)
+ if (bucketizedChildNames != null)
{
for (String bucketizedChildName : bucketizedChildNames)
{
String bucketizedChildPath = childPath + "/" + bucketizedChildName;
subscribeDataChange(bucketizedChildPath, context);
- }
+ }
}
} else
{
@@ -347,13 +374,13 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
default:
{
List<String> childNames = _zkClient.getChildren(path);
- if (childNames != null)
+ if (childNames != null)
{
for (String childName : childNames)
{
String childPath = path + "/" + childName;
subscribeDataChange(childPath, context);
- }
+ }
}
break;
}
@@ -361,7 +388,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
}
catch (ZkNoNodeException e)
{
- logger.warn("fail to subscribe child/data change. path: " + path
+ logger.warn("fail to subscribe child/data change. path: " + path
+ ", listener: " + _listener, e);
}
}
@@ -376,7 +403,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
/**
* Invoke the listener so that it sets up the initial values from the zookeeper if any
* exists
- *
+ *
*/
public void init()
{
@@ -409,7 +436,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
}
catch (Exception e)
{
- String msg = "exception in handling data-change. path: " + dataPath
+ String msg = "exception in handling data-change. path: " + dataPath
+ ", listener: " + _listener;
ZKExceptionHandler.getInstance().handle(msg, e);
}
@@ -423,11 +450,11 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
updateNotificationTime(System.nanoTime());
if (dataPath != null && dataPath.startsWith(_path))
{
- logger.info(_manager.getInstanceName() + " unsubscribe data-change. path: "
+ logger.info(_manager.getInstanceName() + " unsubscribe data-change. path: "
+ dataPath + ", listener: " + _listener);
_zkClient.unsubscribeDataChanges(dataPath, this);
- // only needed for bucketized parent, but OK if we don't have child-change
+ // only needed for bucketized parent, but OK if we don't have child-change
// watch on the bucketized parent path
logger.info(_manager.getInstanceName() + " unsubscribe child-change. path: "
+ dataPath + ", listener: " + _listener);
@@ -440,7 +467,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
}
catch (Exception e)
{
- String msg = "exception in handling data-delete-change. path: " + dataPath
+ String msg = "exception in handling data-delete-change. path: " + dataPath
+ ", listener: " + _listener;
ZKExceptionHandler.getInstance().handle(msg, e);
}
@@ -455,7 +482,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
if (parentPath != null && parentPath.startsWith(_path))
{
NotificationContext changeContext = new NotificationContext(_manager);
-
+
if (currentChilds == null) {
// parentPath has been removed
if (parentPath.equals(_path)) {
@@ -471,7 +498,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
}
catch (Exception e)
{
- String msg = "exception in handling child-change. instance: " + _manager.getInstanceName()
+ String msg = "exception in handling child-change. instance: " + _manager.getInstanceName()
+ ", parentPath: " + parentPath + ", listener: " + _listener;
ZKExceptionHandler.getInstance().handle(msg, e);
}
@@ -479,7 +506,7 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener
/**
* Invoke the listener for the last time so that the listener could clean up resources
- *
+ *
*/
public void reset()
{
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
index fa5100e..4542801 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ControllerManager.java
@@ -134,18 +134,11 @@ public class ControllerManager extends AbstractManager {
}
/**
- * init user defined handlers only
+ * init handlers
+ * ok to init message handler and controller handlers twice
+ * the second init will be skipped (see CallbackHandler)
*/
- List<CallbackHandler> userHandlers = new ArrayList<CallbackHandler>();
- for (CallbackHandler handler : _handlers) {
- Object listener = handler.getListener();
- if (!listener.equals(_messagingService.getExecutor())
- && !listener.equals(_controller)) {
- userHandlers.add(handler);
- }
- }
- initHandlers(userHandlers);
-
+ initHandlers(_handlers);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
index 9017132..2a05a45 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedControllerManager.java
@@ -155,18 +155,11 @@ public class DistributedControllerManager extends AbstractManager {
startTimerTasks();
/**
- * init user defined handlers only
+ * init handlers
+ * ok to init message handler, data-accessor, and controller handlers twice
+ * the second init will be skipped (see CallbackHandler)
*/
- List<CallbackHandler> userHandlers = new ArrayList<CallbackHandler>();
- for (CallbackHandler handler : _handlers) {
- Object listener = handler.getListener();
- if (!listener.equals(_messagingService.getExecutor())
- && !listener.equals(_dataAccessor)
- && !listener.equals(_controller)) {
- userHandlers.add(handler);
- }
- }
- initHandlers(userHandlers);
+ initHandlers(_handlers);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
index ab1da23..bf184cc 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
@@ -24,22 +24,19 @@ import java.lang.management.ManagementFactory;
import org.apache.helix.ControllerChangeListener;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
-import org.apache.helix.HelixManagerFactory;
import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
import org.apache.helix.PropertyType;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.controller.GenericHelixController;
-import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.controller.restlet.ZKPropertyTransferServer;
import org.apache.helix.model.LeaderHistory;
import org.apache.helix.model.LiveInstance;
-import org.apache.helix.participant.DistClusterControllerElection;
import org.apache.log4j.Logger;
/**
* do distributed leader election
- *
+ *
*/
public class DistributedLeaderElection implements ControllerChangeListener {
private static Logger LOG = Logger.getLogger(DistributedLeaderElection.class);
@@ -55,7 +52,7 @@ public class DistributedLeaderElection implements ControllerChangeListener {
/**
* may be accessed by multiple threads: zk-client thread and
* ZkHelixManager.disconnect()->reset() TODO: Refactor accessing
- * HelixMaangerMain class statically
+ * HelixMangerMain class statically
*/
@Override
public synchronized void onControllerChange(NotificationContext changeContext) {
@@ -76,12 +73,17 @@ public class DistributedLeaderElection implements ControllerChangeListener {
try {
if (changeContext.getType().equals(NotificationContext.Type.INIT)
|| changeContext.getType().equals(NotificationContext.Type.CALLBACK)) {
+ LOG.info(_manager.getInstanceName() + " is trying to acquire leadership for cluster: "
+ + _manager.getClusterName());
HelixDataAccessor accessor = manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
while (accessor.getProperty(keyBuilder.controllerLeader()) == null) {
boolean success = tryUpdateController(manager);
if (success) {
+ LOG.info(_manager.getInstanceName() + " acquired leadership for cluster: "
+ + _manager.getClusterName());
+
updateHistory(manager);
_manager._baseDataAccessor.reset();
controllerHelper.addListenersToController(_controller);
@@ -89,9 +91,11 @@ public class DistributedLeaderElection implements ControllerChangeListener {
}
}
} else if (changeContext.getType().equals(NotificationContext.Type.FINALIZE)) {
+ LOG.info(_manager.getInstanceName() + " reqlinquish leadership for cluster: "
+ + _manager.getClusterName());
controllerHelper.stopControllerTimerTasks();
controllerHelper.removeListenersFromController(_controller);
-
+
/**
* clear write-through cache
*/
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
index d175d46..2674598 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
@@ -125,17 +125,11 @@ public class ParticipantManager extends AbstractManager {
startTimerTasks();
/**
- * init user defined handlers only
+ * init handlers
+ * ok to init message handler and data-accessor twice
+ * the second init will be skipped (see CallbackHandler)
*/
- List<CallbackHandler> userHandlers = new ArrayList<CallbackHandler>();
- for (CallbackHandler handler : _handlers) {
- Object listener = handler.getListener();
- if (!listener.equals(_messagingService.getExecutor())
- && !listener.equals(_dataAccessor)) {
- userHandlers.add(handler);
- }
- }
- initHandlers(userHandlers);
+ initHandlers(_handlers);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
index ec6bd79..c9113b7 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManagerHelper.java
@@ -190,7 +190,7 @@ public class ParticipantManagerHelper {
*/
if (retry) {
try {
- _zkclient.createPersistent(liveInstancePath, liveInstance.getRecord());
+ _zkclient.createEphemeral(liveInstancePath, liveInstance.getRecord());
} catch (Exception e) {
String errorMessage = "instance: " + _instanceName
+ " already has a live-instance in cluster " + _clusterName;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/TestHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/TestHelper.java b/helix-core/src/test/java/org/apache/helix/TestHelper.java
index 58dc275..6516138 100644
--- a/helix-core/src/test/java/org/apache/helix/TestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/TestHelper.java
@@ -43,11 +43,15 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.I0Itec.zkclient.IDefaultNameSpace;
+import org.I0Itec.zkclient.IZkChildListener;
+import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkServer;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.apache.commons.io.FileUtils;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.controller.HelixControllerMain;
+import org.apache.helix.integration.manager.ZkTestManager;
+import org.apache.helix.manager.zk.CallbackHandler;
import org.apache.helix.manager.zk.ZKHelixAdmin;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZNRecordSerializer;
@@ -85,7 +89,7 @@ public class TestHelper
return port;
}
-
+
static public ZkServer startZkServer(final String zkAddress) throws Exception
{
List<String> empty = Collections.emptyList();
@@ -103,7 +107,7 @@ public class TestHelper
final List<String> rootNamespaces) throws Exception {
return startZkServer(zkAddress, rootNamespaces, true);
}
-
+
static public ZkServer startZkServer(final String zkAddress,
final List<String> rootNamespaces, boolean overwrite) throws Exception
{
@@ -127,7 +131,7 @@ public class TestHelper
if (rootNamespaces == null) {
return;
}
-
+
for (String rootNamespace : rootNamespaces)
{
try
@@ -210,7 +214,7 @@ public class TestHelper
return manager;
}
-
+
// TODO refactor this
public static StartCMResult startController(final String clusterName,
final String controllerName,
@@ -271,7 +275,7 @@ public class TestHelper
/**
* convert T[] to set<T>
- *
+ *
* @param s
* @return
*/
@@ -288,7 +292,7 @@ public class TestHelper
/**
* generic method for verification with a timeout
- *
+ *
* @param verifierName
* @param args
*/
@@ -471,7 +475,7 @@ public class TestHelper
}
/**
- *
+ *
* @param stateMap
* : "ResourceName/partitionKey" -> setOf(instances)
* @param state
@@ -518,10 +522,10 @@ public class TestHelper
}
/**
- *
+ *
* @param resourcePartition
* : key is in form of "resource/partitionKey" or "resource_x"
- *
+ *
* @return
*/
private static Map<String, String> getResourceAndPartitionKey(String resourcePartition)
@@ -1060,11 +1064,11 @@ public class TestHelper
return sb.toString();
}
-
+
public static interface Verifier {
boolean verify() throws Exception;
}
-
+
public static boolean verify(Verifier verifier, long timeout) throws Exception {
long start = System.currentTimeMillis();
do {
@@ -1075,4 +1079,54 @@ public class TestHelper
Thread.sleep(100);
} while (true);
}
+
+ // debug code
+ public static String printHandlers(ZkTestManager manager)
+ {
+ StringBuilder sb = new StringBuilder();
+ List<CallbackHandler> handlers = manager.getHandlers();
+ sb.append(manager.getInstanceName() + " has " + handlers.size() + " cb-handlers. [");
+
+ for (int i = 0; i < handlers.size(); i++)
+ {
+ CallbackHandler handler = handlers.get(i);
+ String path = handler.getPath();
+ sb.append(path.substring(manager.getClusterName().length() + 1) + ": "
+ + handler.getListener());
+ if (i < (handlers.size() - 1))
+ {
+ sb.append(", ");
+ }
+ }
+ sb.append("]");
+
+ return sb.toString();
+ }
+
+ public static void printZkListeners(ZkClient client) throws Exception{
+ Map<String, Set<IZkDataListener>> datalisteners = ZkTestHelper.getZkDataListener(client);
+ Map<String, Set<IZkChildListener>> childListeners = ZkTestHelper.getZkChildListener(client);
+
+ System.out.println("dataListeners {");
+ for (String path : datalisteners.keySet()) {
+ System.out.println("\t" + path + ": ");
+ Set<IZkDataListener> set = datalisteners.get(path);
+ for (IZkDataListener listener : set) {
+ CallbackHandler handler = (CallbackHandler)listener;
+ System.out.println("\t\t" + handler.getListener());
+ }
+ }
+ System.out.println("}");
+
+ System.out.println("childListeners {");
+ for (String path : childListeners.keySet()) {
+ System.out.println("\t" + path + ": ");
+ Set<IZkChildListener> set = childListeners.get(path);
+ for (IZkChildListener listener : set) {
+ CallbackHandler handler = (CallbackHandler)listener;
+ System.out.println("\t\t" + handler.getListener());
+ }
+ }
+ System.out.println("}");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/integration/manager/ClusterDistributedController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/manager/ClusterDistributedController.java b/helix-core/src/test/java/org/apache/helix/integration/manager/ClusterDistributedController.java
new file mode 100644
index 0000000..56d3aea
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/manager/ClusterDistributedController.java
@@ -0,0 +1,103 @@
+package org.apache.helix.integration.manager;
+
+/*
+ * 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.List;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.helix.manager.zk.CallbackHandler;
+import org.apache.helix.manager.zk.DistributedControllerManager;
+import org.apache.helix.manager.zk.ZkClient;
+import org.apache.log4j.Logger;
+
+public class ClusterDistributedController extends DistributedControllerManager implements Runnable, ZkTestManager
+{
+ private static Logger LOG = Logger.getLogger(ClusterDistributedController.class);
+
+ private final CountDownLatch _startCountDown = new CountDownLatch(1);
+ private final CountDownLatch _stopCountDown = new CountDownLatch(1);
+ private final CountDownLatch _waitStopFinishCountDown = new CountDownLatch(1);
+
+ public ClusterDistributedController(String zkAddr, String clusterName, String controllerName)
+ {
+ super(zkAddr, clusterName, controllerName);
+ }
+
+ public void syncStop()
+ {
+ _stopCountDown.countDown();
+ try
+ {
+ _waitStopFinishCountDown.await();
+ }
+ catch (InterruptedException e)
+ {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ public void syncStart()
+ {
+ // TODO: prevent start multiple times
+ new Thread(this).start();
+ try
+ {
+ _startCountDown.await();
+ }
+ catch (InterruptedException e)
+ {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void run()
+ {
+ try
+ {
+ connect();
+ _startCountDown.countDown();
+ _stopCountDown.await();
+ }
+ catch (Exception e)
+ {
+ LOG.error("exception running controller-manager", e);
+ }
+ finally
+ {
+ _startCountDown.countDown();
+ disconnect();
+ _waitStopFinishCountDown.countDown();
+ }
+ }
+
+ @Override
+ public ZkClient getZkClient() {
+ return _zkclient;
+ }
+
+ @Override
+ public List<CallbackHandler> getHandlers() {
+ return _handlers;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java b/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
index 0f8a27a..8bbb684 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
@@ -21,13 +21,23 @@ package org.apache.helix.integration.manager;
import java.util.Date;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.helix.PreConnectCallback;
+import org.apache.helix.PropertyKey;
import org.apache.helix.TestHelper;
+import org.apache.helix.ZNRecord;
import org.apache.helix.ZkTestHelper;
import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.manager.zk.CallbackHandler;
+import org.apache.helix.manager.zk.DistributedControllerManager;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.mock.participant.MockMSModelFactory;
+import org.apache.helix.model.LiveInstance;
import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
import org.apache.log4j.Logger;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -74,7 +84,7 @@ public class TestConsecutiveZkSessionExpiry extends ZkUnitTestBase
}
@Test
- public void test() throws Exception
+ public void testParticipant() throws Exception
{
// Logger.getRootLogger().setLevel(Level.INFO);
String className = TestHelper.getTestClassName();
@@ -99,7 +109,7 @@ public class TestConsecutiveZkSessionExpiry extends ZkUnitTestBase
// start controller
final ClusterControllerManager controller =
new ClusterControllerManager(ZK_ADDR, clusterName, "controller");
- controller.connect();
+ controller.syncStart();
// start participants
CountDownLatch startCountdown = new CountDownLatch(1);
@@ -152,8 +162,106 @@ public class TestConsecutiveZkSessionExpiry extends ZkUnitTestBase
clusterName));
Assert.assertTrue(result);
+ // clean up
+ controller.syncStop();
+ for (int i = 0; i < n; i++) {
+ participants[i].syncStop();
+ }
+
System.out.println("END " + clusterName + " at "
+ new Date(System.currentTimeMillis()));
}
+ @Test
+ public void testDistributedController() throws Exception {
+ // Logger.getRootLogger().setLevel(Level.INFO);
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+ int n = 2;
+
+ System.out.println("START " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 4, // partitions per resource
+ n, // number of nodes
+ 2, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+ ClusterDistributedController[] distributedControllers = new ClusterDistributedController[n];
+ CountDownLatch startCountdown = new CountDownLatch(1);
+ CountDownLatch endCountdown = new CountDownLatch(1);
+
+ for (int i = 0; i < n; i++) {
+ String contrllerName = "localhost_" + (12918 + i);
+ distributedControllers[i] = new ClusterDistributedController(ZK_ADDR, clusterName, contrllerName);
+ distributedControllers[i].getStateMachineEngine().registerStateModelFactory("MasterSlave", new MockMSModelFactory());
+ if (i == 0) {
+ distributedControllers[i].addPreConnectCallback(new PreConnectTestCallback(contrllerName,
+ startCountdown,
+ endCountdown));
+ }
+ distributedControllers[i].connect();
+ }
+
+ boolean result =
+ ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+
+ // expire the session of distributedController
+ LOG.info("1st Expiring distributedController session...");
+ String oldSessionId = distributedControllers[0].getSessionId();
+
+ ZkTestHelper.asyncExpireSession(distributedControllers[0].getZkClient());
+ String newSessionId = distributedControllers[0].getSessionId();
+ LOG.info("Expried distributedController session. oldSessionId: " + oldSessionId
+ + ", newSessionId: " + newSessionId);
+
+ // expire zk session again during HelixManager#handleNewSession()
+ startCountdown.await();
+ LOG.info("2nd Expiring distributedController session...");
+ oldSessionId = distributedControllers[0].getSessionId();
+
+ ZkTestHelper.asyncExpireSession(distributedControllers[0].getZkClient());
+ newSessionId = distributedControllers[0].getSessionId();
+ LOG.info("Expried distributedController session. oldSessionId: " + oldSessionId
+ + ", newSessionId: " + newSessionId);
+
+ endCountdown.countDown();
+
+ result =
+ ClusterStateVerifier.verifyByPolling(new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+
+ // verify leader changes to localhost_12919
+ ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Assert.assertNotNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12918")));
+ LiveInstance leader = accessor.getProperty(keyBuilder.controllerLeader());
+ Assert.assertNotNull(leader);
+ Assert.assertEquals(leader.getId(), "localhost_12919");
+
+ // check localhost_12918 has 2 handlers: message and data-accessor
+ LOG.debug("handlers: " + TestHelper.printHandlers(distributedControllers[0]));
+ List<CallbackHandler> handlers = distributedControllers[0].getHandlers();
+ Assert.assertEquals(handlers.size(), 2,
+ "Distributed controller should have 2 handlers (message and data-accessor) after lose leadership, but was "
+ + handlers.size());
+
+ // clean up
+ distributedControllers[1].disconnect();
+ Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12919")));
+ Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));
+
+
+ System.out.println("END " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/integration/manager/TestControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/manager/TestControllerManager.java b/helix-core/src/test/java/org/apache/helix/integration/manager/TestControllerManager.java
new file mode 100644
index 0000000..07d019d
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/manager/TestControllerManager.java
@@ -0,0 +1,174 @@
+package org.apache.helix.integration.manager;
+
+/*
+ * 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.Date;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZkTestHelper;
+import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class TestControllerManager extends ZkUnitTestBase
+{
+
+ @Test
+ public void testMultipleControllersOfSameName() throws Exception {
+ // Logger.getRootLogger().setLevel(Level.INFO);
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+ int n = 5;
+
+ System.out.println("START " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 16, // partitions per resource
+ n, // number of nodes
+ 3, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+ // start multiple controllers of same name
+ int m = 3;
+ ClusterControllerManager[] controllers = new ClusterControllerManager[m];
+
+ for (int i = 0; i < m; i++) {
+ controllers[i] = new ClusterControllerManager(ZK_ADDR, clusterName, "controller");
+ controllers[i].syncStart();
+ }
+
+ // assert that only one is leader
+ int leaderCnt = 0;
+ for (int i = 0; i < m; i++) {
+ if (controllers[i].isLeader()) {
+ leaderCnt++;
+ }
+ }
+ Assert.assertEquals(leaderCnt, 1, "Should have only 1 leader but was " + leaderCnt);
+
+ MockParticipantManager[] participants = new MockParticipantManager[n];
+ for (int i = 0; i < n; i++)
+ {
+ final String instanceName = "localhost_" + (12918 + i);
+
+ participants[i] =
+ new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+
+ participants[i].syncStart();
+ }
+
+ boolean result =
+ ClusterStateVerifier.verifyByZkCallback(new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+
+ // cleanup
+ for (int i = 0; i < m; i++) {
+ controllers[i].syncStop();
+ }
+
+ for (int i = 0; i < n; i++) {
+ participants[i].syncStop();
+ }
+
+ System.out.println("END " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+ }
+
+ @Test
+ public void simpleSessionExpiryTest() throws Exception
+ {
+// Logger.getRootLogger().setLevel(Level.WARN);
+
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ final String clusterName = className + "_" + methodName;
+ int n = 5;
+
+ System.out.println("START " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+
+ MockParticipantManager[] participants = new MockParticipantManager[n];
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 16, // partitions per resource
+ n, // number of nodes
+ 3, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+ // start controller
+ ClusterControllerManager controller = new ClusterControllerManager(ZK_ADDR, clusterName, "controller");
+ controller.syncStart();
+
+ // start participants
+ for (int i = 0; i < n; i++)
+ {
+ String instanceName = "localhost_" + (12918 + i);
+ participants[i] =
+ new MockParticipantManager(ZK_ADDR,
+ clusterName,
+ instanceName);
+ participants[i].syncStart();
+ }
+
+ boolean result =
+ ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+ String oldSessionId = controller.getSessionId();
+
+ // expire zk-connection on localhost_12918
+ ZkTestHelper.expireSession(controller.getZkClient());
+
+ // wait until session expiry callback happens
+ TimeUnit.MILLISECONDS.sleep(100);
+
+ result =
+ ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+ String newSessionId = controller.getSessionId();
+ Assert.assertNotSame(newSessionId, oldSessionId);
+
+ // cleanup
+ controller.syncStop();
+ for (int i = 0; i < n; i++) {
+ participants[i].syncStop();
+ }
+
+ System.out.println("END " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/integration/manager/TestDistributedControllerManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/manager/TestDistributedControllerManager.java b/helix-core/src/test/java/org/apache/helix/integration/manager/TestDistributedControllerManager.java
index 319633b..6af0380 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/manager/TestDistributedControllerManager.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/manager/TestDistributedControllerManager.java
@@ -20,24 +20,27 @@ package org.apache.helix.integration.manager;
*/
import java.util.Date;
+import java.util.List;
import org.apache.helix.PropertyKey;
import org.apache.helix.TestHelper;
import org.apache.helix.ZNRecord;
+import org.apache.helix.ZkTestHelper;
import org.apache.helix.integration.ZkIntegrationTestBase;
-import org.apache.helix.manager.zk.ControllerManager;
+import org.apache.helix.manager.zk.CallbackHandler;
import org.apache.helix.manager.zk.DistributedControllerManager;
-import org.apache.helix.manager.zk.ParticipantManager;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.mock.participant.MockMSModelFactory;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.tools.ClusterStateVerifier;
import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
+import org.apache.log4j.Logger;
import org.testng.Assert;
import org.testng.annotations.Test;
public class TestDistributedControllerManager extends ZkIntegrationTestBase {
+ private static Logger LOG = Logger.getLogger(TestDistributedControllerManager.class);
@Test
public void simpleIntegrationTest() throws Exception {
@@ -59,7 +62,7 @@ public class TestDistributedControllerManager extends ZkIntegrationTestBase {
2, // replicas
"MasterSlave",
true); // do rebalance
-
+
DistributedControllerManager[] distributedControllers = new DistributedControllerManager[n];
for (int i = 0; i < n; i++) {
int port = 12918 + i;
@@ -67,7 +70,7 @@ public class TestDistributedControllerManager extends ZkIntegrationTestBase {
distributedControllers[i].getStateMachineEngine().registerStateModelFactory("MasterSlave", new MockMSModelFactory());
distributedControllers[i].connect();
}
-
+
boolean result =
ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
clusterName));
@@ -75,14 +78,14 @@ public class TestDistributedControllerManager extends ZkIntegrationTestBase {
// disconnect first distributed-controller, and verify second takes leadership
distributedControllers[0].disconnect();
-
+
// verify leader changes to localhost_12919
Thread.sleep(100);
result =
ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
clusterName));
Assert.assertTrue(result);
-
+
ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12918")));
@@ -95,8 +98,108 @@ public class TestDistributedControllerManager extends ZkIntegrationTestBase {
Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12919")));
Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));
-
+
System.out.println("START " + clusterName + " at "
+ new Date(System.currentTimeMillis()));
}
+
+ /**
+ * expire a controller and make sure the other takes the leadership
+ *
+ * @param expireController
+ * @param newController
+ * @throws Exception
+ */
+ void expireController(ClusterDistributedController expireController,
+ ClusterDistributedController newController) throws Exception
+ {
+ String clusterName = expireController.getClusterName();
+ LOG.info("Expiring distributedController: " + expireController.getInstanceName()
+ + ", session: " + expireController.getSessionId() + " ...");
+ String oldSessionId = expireController.getSessionId();
+
+ ZkTestHelper.expireSession(expireController.getZkClient());
+ String newSessionId = expireController.getSessionId();
+ LOG.debug("Expried distributedController: " + expireController.getInstanceName()
+ + ", oldSessionId: " + oldSessionId + ", newSessionId: " + newSessionId);
+
+ boolean result =
+ ClusterStateVerifier.verifyByPolling(new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+
+ // verify leader changes to localhost_12919
+ ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Assert.assertNotNull(accessor.getProperty(keyBuilder.liveInstance(expireController.getInstanceName())));
+ LiveInstance leader = accessor.getProperty(keyBuilder.controllerLeader());
+ Assert.assertNotNull(leader);
+ Assert.assertEquals(leader.getId(), newController.getInstanceName());
+
+
+ // check expired-controller has 2 handlers: message and data-accessor
+ LOG.debug(expireController.getInstanceName() + " handlers: " + TestHelper.printHandlers(expireController));
+
+ List<CallbackHandler> handlers = expireController.getHandlers();
+ Assert.assertEquals(handlers.size(), 2,
+ "Distributed controller should have 2 handlers (message and data-accessor) after lose leadership, but was "
+ + handlers.size());
+ }
+
+ @Test
+ public void simpleSessionExpiryTest() throws Exception {
+ // Logger.getRootLogger().setLevel(Level.INFO);
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+ int n = 2;
+
+ System.out.println("START " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+
+ TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 4, // partitions per resource
+ n, // number of nodes
+ 2, // replicas
+ "MasterSlave",
+ true); // do rebalance
+
+ ClusterDistributedController[] distributedControllers = new ClusterDistributedController[n];
+
+ for (int i = 0; i < n; i++) {
+ String contrllerName = "localhost_" + (12918 + i);
+ distributedControllers[i] = new ClusterDistributedController(ZK_ADDR, clusterName, contrllerName);
+ distributedControllers[i].getStateMachineEngine().registerStateModelFactory("MasterSlave", new MockMSModelFactory());
+ distributedControllers[i].connect();
+ }
+
+ boolean result =
+ ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
+ clusterName));
+ Assert.assertTrue(result);
+
+ // expire localhost_12918
+ expireController(distributedControllers[0], distributedControllers[1]);
+
+ // expire localhost_12919
+ expireController(distributedControllers[1], distributedControllers[0]);
+
+
+ // clean up
+ ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ for (int i = 0; i < n; i++) {
+ distributedControllers[i].disconnect();
+ Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance(distributedControllers[i].getInstanceName())));
+ }
+ Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));
+
+
+ System.out.println("END " + clusterName + " at "
+ + new Date(System.currentTimeMillis()));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/6d539799/helix-core/src/test/java/org/apache/helix/integration/manager/TestParticipantManager.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/manager/TestParticipantManager.java b/helix-core/src/test/java/org/apache/helix/integration/manager/TestParticipantManager.java
index a0012f2..3cfdb76 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/manager/TestParticipantManager.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/manager/TestParticipantManager.java
@@ -94,7 +94,7 @@ public class TestParticipantManager extends ZkIntegrationTestBase {
Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12918")));
Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));
- System.out.println("START " + clusterName + " at "
+ System.out.println("END " + clusterName + " at "
+ new Date(System.currentTimeMillis()));
}