@Test public void testRestartMaster() throws IOException, KeeperException { ZooKeeperWatcher zk = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(), "testActiveMasterManagerFromZK", null); ZKUtil.createAndFailSilent(zk, zk.baseZNode); try { ZKUtil.deleteNode(zk, zk.masterAddressZNode); } catch(KeeperException.NoNodeException nne) {} HServerAddress master = new HServerAddress("localhost", 1); DummyMaster dummyMaster = new DummyMaster(); ActiveMasterManager activeMasterManager = new ActiveMasterManager(zk, master, dummyMaster); zk.registerListener(activeMasterManager); assertFalse(activeMasterManager.clusterHasActiveMaster.get()); activeMasterManager.blockUntilBecomingActiveMaster(); assertTrue(activeMasterManager.clusterHasActiveMaster.get()); assertMaster(zk, master); DummyMaster secondDummyMaster = new DummyMaster(); ActiveMasterManager secondActiveMasterManager = new ActiveMasterManager(zk, master, secondDummyMaster); zk.registerListener(secondActiveMasterManager); assertFalse(secondActiveMasterManager.clusterHasActiveMaster.get()); activeMasterManager.blockUntilBecomingActiveMaster(); assertTrue(activeMasterManager.clusterHasActiveMaster.get()); assertMaster(zk, master); }
boolean blockUntilBecomingActiveMaster() { boolean cleanSetOfActiveMaster = true; try { if (ZKUtil.setAddressAndWatch(this.watcher, this.watcher.masterAddressZNode, this.address)) { this.clusterHasActiveMaster.set(true); LOG.info("Master=" + this.address); return cleanSetOfActiveMaster; } cleanSetOfActiveMaster = false; this.clusterHasActiveMaster.set(true); HServerAddress currentMaster = ZKUtil.getDataAsAddress(this.watcher, this.watcher.masterAddressZNode); if (currentMaster != null && currentMaster.equals(this.address)) { LOG.info("Current master has this master's address, " + currentMaster + "; master was restarted? Waiting on znode to expire..."); ZKUtil.deleteNode(this.watcher, this.watcher.masterAddressZNode); } else { LOG.info("Another master is the active master, " + currentMaster + "; waiting to become the next active master"); } } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (this.clusterHasActiveMaster.get() && !this.master.isStopped()) { try { this.clusterHasActiveMaster.wait(); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (this.master.isStopped()) { return cleanSetOfActiveMaster; } blockUntilBecomingActiveMaster(); } return cleanSetOfActiveMaster; }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster() { boolean cleanSetOfActiveMaster = true; try { if (ZKUtil.setAddressAndWatch(this.watcher, this.watcher.masterAddressZNode, this.address)) { this.clusterHasActiveMaster.set(true); LOG.info("Master=" + this.address); return cleanSetOfActiveMaster; } cleanSetOfActiveMaster = false; this.clusterHasActiveMaster.set(true); HServerAddress currentMaster = ZKUtil.getDataAsAddress(this.watcher, this.watcher.masterAddressZNode); if (currentMaster != null && currentMaster.equals(this.address)) { LOG.info("Current master has this master's address, " + currentMaster + "; master was restarted? Waiting on znode to expire..."); ZKUtil.deleteNode(this.watcher, this.watcher.masterAddressZNode); } else { LOG.info("Another master is the active master, " + currentMaster + "; waiting to become the next active master"); } } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (this.clusterHasActiveMaster.get() && !this.master.isStopped()) { try { this.clusterHasActiveMaster.wait(); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (this.master.isStopped()) { return cleanSetOfActiveMaster; } blockUntilBecomingActiveMaster(); } return cleanSetOfActiveMaster; } }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster() { boolean cleanSetOfActiveMaster = true; try { if (ZKUtil.setAddressAndWatch(this.watcher, this.watcher.masterAddressZNode, this.address)) { this.clusterHasActiveMaster.set(true); LOG.info("Master=" + this.address); return cleanSetOfActiveMaster; } cleanSetOfActiveMaster = false; this.clusterHasActiveMaster.set(true); HServerAddress currentMaster = ZKUtil.getDataAsAddress(this.watcher, this.watcher.masterAddressZNode); if (currentMaster != null && currentMaster.equals(this.address)) { LOG.info("Current master has this master's address, " + currentMaster + "; master was restarted? Waiting on znode to expire..."); ZKUtil.deleteNode(this.watcher, this.watcher.masterAddressZNode); } else { LOG.info("Another master is the active master, " + currentMaster + "; waiting to become the next active master"); } } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (this.clusterHasActiveMaster.get() && !this.master.isStopped()) { try { this.clusterHasActiveMaster.wait(); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (this.master.isStopped()) { return cleanSetOfActiveMaster; } blockUntilBecomingActiveMaster(); } return cleanSetOfActiveMaster; } ActiveMasterManager(ZooKeeperWatcher watcher, HServerAddress address,
      Server master); }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster() { boolean cleanSetOfActiveMaster = true; try { if (ZKUtil.setAddressAndWatch(this.watcher, this.watcher.masterAddressZNode, this.address)) { this.clusterHasActiveMaster.set(true); LOG.info("Master=" + this.address); return cleanSetOfActiveMaster; } cleanSetOfActiveMaster = false; this.clusterHasActiveMaster.set(true); HServerAddress currentMaster = ZKUtil.getDataAsAddress(this.watcher, this.watcher.masterAddressZNode); if (currentMaster != null && currentMaster.equals(this.address)) { LOG.info("Current master has this master's address, " + currentMaster + "; master was restarted? Waiting on znode to expire..."); ZKUtil.deleteNode(this.watcher, this.watcher.masterAddressZNode); } else { LOG.info("Another master is the active master, " + currentMaster + "; waiting to become the next active master"); } } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (this.clusterHasActiveMaster.get() && !this.master.isStopped()) { try { this.clusterHasActiveMaster.wait(); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (this.master.isStopped()) { return cleanSetOfActiveMaster; } blockUntilBecomingActiveMaster(); } return cleanSetOfActiveMaster; } ActiveMasterManager(ZooKeeperWatcher watcher, HServerAddress address,
      Server master); @Override void nodeCreated(String path); @Override void nodeDeleted(String path); boolean isActiveMaster(); void stop(); }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster() { boolean cleanSetOfActiveMaster = true; try { if (ZKUtil.setAddressAndWatch(this.watcher, this.watcher.masterAddressZNode, this.address)) { this.clusterHasActiveMaster.set(true); LOG.info("Master=" + this.address); return cleanSetOfActiveMaster; } cleanSetOfActiveMaster = false; this.clusterHasActiveMaster.set(true); HServerAddress currentMaster = ZKUtil.getDataAsAddress(this.watcher, this.watcher.masterAddressZNode); if (currentMaster != null && currentMaster.equals(this.address)) { LOG.info("Current master has this master's address, " + currentMaster + "; master was restarted? Waiting on znode to expire..."); ZKUtil.deleteNode(this.watcher, this.watcher.masterAddressZNode); } else { LOG.info("Another master is the active master, " + currentMaster + "; waiting to become the next active master"); } } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (this.clusterHasActiveMaster.get() && !this.master.isStopped()) { try { this.clusterHasActiveMaster.wait(); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (this.master.isStopped()) { return cleanSetOfActiveMaster; } blockUntilBecomingActiveMaster(); } return cleanSetOfActiveMaster; } ActiveMasterManager(ZooKeeperWatcher watcher, HServerAddress address,
      Server master); @Override void nodeCreated(String path); @Override void nodeDeleted(String path); boolean isActiveMaster(); void stop();  }