@Test public void testHandleMasterNodeChange() throws IOException, KeeperException, InterruptedException { HostPort hostPort1 = new HostPort("127.0.0.1", 11111); FailoverWatcher failoverWatcher1 = createFailoverWatcher(hostPort1); failoverWatcher1.blockUntilActive(); HostPort hostPort2 = new HostPort("127.0.0.1", 22222); final FailoverWatcher failoverWatcher2 = createFailoverWatcher(hostPort2); Thread thread2 = new Thread() { @Override public void run() { failoverWatcher2.blockUntilActive(); } }; thread2.start(); HostPort hostPort3 = new HostPort("127.0.0.1", 33333); final FailoverWatcher failoverWatcher3 = createFailoverWatcher(hostPort3); Thread thread3 = new Thread() { @Override public void run() { failoverWatcher3.blockUntilActive(); } }; thread3.start(); ZooKeeperUtil.deleteNode(failoverWatcher1, failoverWatcher1.masterZnode); Thread.sleep(500); String masterZnodeData = new String(ZooKeeperUtil.getDataAndWatch(failoverWatcher1, failoverWatcher1.masterZnode)); String backupMasterNode = ZooKeeperUtil.listChildrenAndWatchForNewChildren(failoverWatcher1, failoverWatcher1.backupServersZnode).get(0); if (masterZnodeData.equals(hostPort2.getHostPort())) { if (!backupMasterNode.equals(hostPort3.getHostPort())) { assert false; } } else if (masterZnodeData.equals(hostPort3.getHostPort())) { if (!backupMasterNode.equals(hostPort2.getHostPort())) { assert false; } } else { assert false; } if (thread2.isAlive()) { thread2.interrupt(); } else { thread3.interrupt(); } failoverWatcher1.close(); failoverWatcher2.close(); }
private void handleMasterNodeChange() { try { synchronized (hasActiveServer) { if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) { if (LOG.isDebugEnabled()) { LOG.debug("A master is now available"); } hasActiveServer.set(true); } else { if (LOG.isDebugEnabled()) { LOG.debug("No master available. Notifying waiting threads"); } hasActiveServer.set(false); hasActiveServer.notifyAll(); } } } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException, aborting", ke); } }
FailoverWatcher implements Watcher { private void handleMasterNodeChange() { try { synchronized (hasActiveServer) { if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) { if (LOG.isDebugEnabled()) { LOG.debug("A master is now available"); } hasActiveServer.set(true); } else { if (LOG.isDebugEnabled()) { LOG.debug("No master available. Notifying waiting threads"); } hasActiveServer.set(false); hasActiveServer.notifyAll(); } } } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException, aborting", ke); } } }
FailoverWatcher implements Watcher { private void handleMasterNodeChange() { try { synchronized (hasActiveServer) { if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) { if (LOG.isDebugEnabled()) { LOG.debug("A master is now available"); } hasActiveServer.set(true); } else { if (LOG.isDebugEnabled()) { LOG.debug("No master available. Notifying waiting threads"); } hasActiveServer.set(false); hasActiveServer.notifyAll(); } } } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException, aborting", ke); } } FailoverWatcher(Properties properties, boolean canInitZnode);  FailoverWatcher(Properties properties); }
FailoverWatcher implements Watcher { private void handleMasterNodeChange() { try { synchronized (hasActiveServer) { if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) { if (LOG.isDebugEnabled()) { LOG.debug("A master is now available"); } hasActiveServer.set(true); } else { if (LOG.isDebugEnabled()) { LOG.debug("No master available. Notifying waiting threads"); } hasActiveServer.set(false); hasActiveServer.notifyAll(); } } } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException, aborting", ke); } } FailoverWatcher(Properties properties, boolean canInitZnode);  FailoverWatcher(Properties properties); @Override void process(WatchedEvent event); boolean blockUntilActive(); void close(); void waitToInitZooKeeper(long maxWaitMillis); String getBaseZnode(); String getMasterZnode(); String getBackupServersZnode(); ZooKeeper getZooKeeper(); boolean hasActiveServer(); HostPort getHostPort(); Properties getProperties(); boolean isZkSecure(); String getZkAdmin(); }
FailoverWatcher implements Watcher { private void handleMasterNodeChange() { try { synchronized (hasActiveServer) { if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) { if (LOG.isDebugEnabled()) { LOG.debug("A master is now available"); } hasActiveServer.set(true); } else { if (LOG.isDebugEnabled()) { LOG.debug("No master available. Notifying waiting threads"); } hasActiveServer.set(false); hasActiveServer.notifyAll(); } } } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException, aborting", ke); } } FailoverWatcher(Properties properties, boolean canInitZnode);  FailoverWatcher(Properties properties); @Override void process(WatchedEvent event); boolean blockUntilActive(); void close(); void waitToInitZooKeeper(long maxWaitMillis); String getBaseZnode(); String getMasterZnode(); String getBackupServersZnode(); ZooKeeper getZooKeeper(); boolean hasActiveServer(); HostPort getHostPort(); Properties getProperties(); boolean isZkSecure(); String getZkAdmin();  }