@Test public void testBlockUntilActive() throws IOException, KeeperException, InterruptedException { HostPort hostPort1 = new HostPort("127.0.0.1", 11111); FailoverWatcher failoverWatcher1 = createFailoverWatcher(hostPort1); failoverWatcher1.blockUntilActive(); assertTrue(failoverWatcher1.hasActiveServer()); String activeServer = new String(ZooKeeperUtil.getDataAndWatch(failoverWatcher1, failoverWatcher1.masterZnode)); assertTrue(activeServer.equals(hostPort1.getHostPort())); 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(); Thread.sleep(500); List<String> backupMastersList = ZooKeeperUtil.listChildrenAndWatchForNewChildren( failoverWatcher1, failoverWatcher1.backupServersZnode); assertTrue(failoverWatcher2.hasActiveServer()); assertTrue(backupMastersList.contains(failoverWatcher2.getHostPort().getHostPort())); thread2.interrupt(); failoverWatcher1.close(); failoverWatcher2.close(); }
public boolean blockUntilActive() { while (true) { try { if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort() .getBytes())) { LOG.info("Deleting ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " from backup master directory"); ZooKeeperUtil.deleteNodeFailSilent(this, backupServersZnode + "/" + hostPort.getHostPort()); hasActiveServer.set(true); LOG.info("Become active master in " + hostPort.getHostPort()); return true; } hasActiveServer.set(true); LOG.info("Adding ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " in backup master directory"); ZooKeeperUtil.createEphemeralNodeAndWatch(this, backupServersZnode + "/" + hostPort.getHostPort(), hostPort.getHostPort().getBytes()); String msg; byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode); if (bytes == null) { msg = ("A master was detected, but went down before its address " + "could be read. Attempting to become the next active master"); } else { if (hostPort.getHostPort().equals(new String(bytes))) { msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node."); ZooKeeperUtil.deleteNode(this, masterZnode); } else { msg = "Another master " + new String(bytes) + " is the active master, " + hostPort.getHostPort() + "; waiting to become the next active master"; } } LOG.info(msg); } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException when block to become active, aborting", ke); return false; } synchronized (hasActiveServer) { while (hasActiveServer.get()) { try { hasActiveServer.wait(); } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Interrupted while waiting to be master"); } return false; } } } } }
FailoverWatcher implements Watcher { public boolean blockUntilActive() { while (true) { try { if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort() .getBytes())) { LOG.info("Deleting ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " from backup master directory"); ZooKeeperUtil.deleteNodeFailSilent(this, backupServersZnode + "/" + hostPort.getHostPort()); hasActiveServer.set(true); LOG.info("Become active master in " + hostPort.getHostPort()); return true; } hasActiveServer.set(true); LOG.info("Adding ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " in backup master directory"); ZooKeeperUtil.createEphemeralNodeAndWatch(this, backupServersZnode + "/" + hostPort.getHostPort(), hostPort.getHostPort().getBytes()); String msg; byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode); if (bytes == null) { msg = ("A master was detected, but went down before its address " + "could be read. Attempting to become the next active master"); } else { if (hostPort.getHostPort().equals(new String(bytes))) { msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node."); ZooKeeperUtil.deleteNode(this, masterZnode); } else { msg = "Another master " + new String(bytes) + " is the active master, " + hostPort.getHostPort() + "; waiting to become the next active master"; } } LOG.info(msg); } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException when block to become active, aborting", ke); return false; } synchronized (hasActiveServer) { while (hasActiveServer.get()) { try { hasActiveServer.wait(); } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Interrupted while waiting to be master"); } return false; } } } } } }
FailoverWatcher implements Watcher { public boolean blockUntilActive() { while (true) { try { if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort() .getBytes())) { LOG.info("Deleting ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " from backup master directory"); ZooKeeperUtil.deleteNodeFailSilent(this, backupServersZnode + "/" + hostPort.getHostPort()); hasActiveServer.set(true); LOG.info("Become active master in " + hostPort.getHostPort()); return true; } hasActiveServer.set(true); LOG.info("Adding ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " in backup master directory"); ZooKeeperUtil.createEphemeralNodeAndWatch(this, backupServersZnode + "/" + hostPort.getHostPort(), hostPort.getHostPort().getBytes()); String msg; byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode); if (bytes == null) { msg = ("A master was detected, but went down before its address " + "could be read. Attempting to become the next active master"); } else { if (hostPort.getHostPort().equals(new String(bytes))) { msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node."); ZooKeeperUtil.deleteNode(this, masterZnode); } else { msg = "Another master " + new String(bytes) + " is the active master, " + hostPort.getHostPort() + "; waiting to become the next active master"; } } LOG.info(msg); } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException when block to become active, aborting", ke); return false; } synchronized (hasActiveServer) { while (hasActiveServer.get()) { try { hasActiveServer.wait(); } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Interrupted while waiting to be master"); } return false; } } } } } FailoverWatcher(Properties properties, boolean canInitZnode);  FailoverWatcher(Properties properties); }
FailoverWatcher implements Watcher { public boolean blockUntilActive() { while (true) { try { if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort() .getBytes())) { LOG.info("Deleting ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " from backup master directory"); ZooKeeperUtil.deleteNodeFailSilent(this, backupServersZnode + "/" + hostPort.getHostPort()); hasActiveServer.set(true); LOG.info("Become active master in " + hostPort.getHostPort()); return true; } hasActiveServer.set(true); LOG.info("Adding ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " in backup master directory"); ZooKeeperUtil.createEphemeralNodeAndWatch(this, backupServersZnode + "/" + hostPort.getHostPort(), hostPort.getHostPort().getBytes()); String msg; byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode); if (bytes == null) { msg = ("A master was detected, but went down before its address " + "could be read. Attempting to become the next active master"); } else { if (hostPort.getHostPort().equals(new String(bytes))) { msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node."); ZooKeeperUtil.deleteNode(this, masterZnode); } else { msg = "Another master " + new String(bytes) + " is the active master, " + hostPort.getHostPort() + "; waiting to become the next active master"; } } LOG.info(msg); } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException when block to become active, aborting", ke); return false; } synchronized (hasActiveServer) { while (hasActiveServer.get()) { try { hasActiveServer.wait(); } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Interrupted while waiting to be master"); } return false; } } } } } 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 { public boolean blockUntilActive() { while (true) { try { if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort() .getBytes())) { LOG.info("Deleting ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " from backup master directory"); ZooKeeperUtil.deleteNodeFailSilent(this, backupServersZnode + "/" + hostPort.getHostPort()); hasActiveServer.set(true); LOG.info("Become active master in " + hostPort.getHostPort()); return true; } hasActiveServer.set(true); LOG.info("Adding ZNode for " + backupServersZnode + "/" + hostPort.getHostPort() + " in backup master directory"); ZooKeeperUtil.createEphemeralNodeAndWatch(this, backupServersZnode + "/" + hostPort.getHostPort(), hostPort.getHostPort().getBytes()); String msg; byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode); if (bytes == null) { msg = ("A master was detected, but went down before its address " + "could be read. Attempting to become the next active master"); } else { if (hostPort.getHostPort().equals(new String(bytes))) { msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node."); ZooKeeperUtil.deleteNode(this, masterZnode); } else { msg = "Another master " + new String(bytes) + " is the active master, " + hostPort.getHostPort() + "; waiting to become the next active master"; } } LOG.info(msg); } catch (KeeperException ke) { LOG.error("Received an unexpected KeeperException when block to become active, aborting", ke); return false; } synchronized (hasActiveServer) { while (hasActiveServer.get()) { try { hasActiveServer.wait(); } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Interrupted while waiting to be master"); } return false; } } } } } 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();  }