ActiveStandbyElector implements StatCallback, StringCallback { synchronized void processWatchEvent(ZooKeeper zk, WatchedEvent event) { Event.EventType eventType = event.getType(); if (isStaleClient(zk)) return; LOG.debug("Watcher event type: " + eventType + " with state:" + event.getState() + " for path:" + event.getPath() + " connectionState: " + zkConnectionState + " for " + this); if (eventType == Event.EventType.None) { switch (event.getState()) { case SyncConnected: LOG.info("Session connected."); ConnectionState prevConnectionState = zkConnectionState; zkConnectionState = ConnectionState.CONNECTED; if (prevConnectionState == ConnectionState.DISCONNECTED && wantToBeInElection) { monitorActiveStatus(); } break; case Disconnected: LOG.info("Session disconnected. Entering neutral mode..."); zkConnectionState = ConnectionState.DISCONNECTED; enterNeutralMode(); break; case Expired: LOG.info("Session expired. Entering neutral mode and rejoining..."); enterNeutralMode(); reJoinElection(0); break; case SaslAuthenticated: LOG.info("Successfully authenticated to ZooKeeper using SASL."); break; default: fatalError("Unexpected Zookeeper watch event state: " + event.getState()); break; } return; } String path = event.getPath(); if (path != null) { switch (eventType) { case NodeDeleted: if (state == State.ACTIVE) { enterNeutralMode(); } joinElectionInternal(); break; case NodeDataChanged: monitorActiveStatus(); break; default: LOG.debug("Unexpected node event: " + eventType + " for path: " + path); monitorActiveStatus(); } return; } fatalError("Unexpected watch error from Zookeeper"); } ActiveStandbyElector(String zookeeperHostPorts,
      int zookeeperSessionTimeout, String parentZnodeName, List<ACL> acl,
      List<ZKAuthInfo> authInfo,
      ActiveStandbyElectorCallback app); synchronized void joinElection(byte[] data); synchronized boolean parentZNodeExists(); synchronized void ensureParentZNode(); synchronized void clearParentZNode(); synchronized void quitElection(boolean needFence); synchronized byte[] getActiveData(); @Override synchronized void processResult(int rc, String path, Object ctx,
      String name); @Override synchronized void processResult(int rc, String path, Object ctx,
      Stat stat); @InterfaceAudience.Private synchronized void terminateConnection(); @Override String toString(); static final Log LOG; }
@Test public void testBecomeActiveBeforeServiceHealthy() throws Exception { mockNoPriorActive(); WatchedEvent mockEvent = Mockito.mock(WatchedEvent.class); Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.None); Mockito.when(mockEvent.getState()).thenReturn(Event.KeeperState.Expired); elector.processWatchEvent(mockZK, mockEvent); Mockito.verify(mockZK, Mockito.times(0)).create(ZK_LOCK_NAME, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK); }