@Test public void testRestartMaster() throws IOException, KeeperException { ZooKeeperWatcher zk = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(), "testActiveMasterManagerFromZK", null, true); try { ZKUtil.deleteNode(zk, zk.getMasterAddressZNode()); ZKUtil.deleteNode(zk, zk.clusterStateZNode); } catch(KeeperException.NoNodeException nne) {} ServerName master = ServerName.valueOf("localhost", 1, System.currentTimeMillis()); DummyMaster dummyMaster = new DummyMaster(zk,master); ClusterStatusTracker clusterStatusTracker = dummyMaster.getClusterStatusTracker(); ActiveMasterManager activeMasterManager = dummyMaster.getActiveMasterManager(); assertFalse(activeMasterManager.clusterHasActiveMaster.get()); MonitoredTask status = Mockito.mock(MonitoredTask.class); clusterStatusTracker.setClusterUp(); activeMasterManager.blockUntilBecomingActiveMaster(100, status); assertTrue(activeMasterManager.clusterHasActiveMaster.get()); assertMaster(zk, master); DummyMaster secondDummyMaster = new DummyMaster(zk,master); ActiveMasterManager secondActiveMasterManager = secondDummyMaster.getActiveMasterManager(); assertFalse(secondActiveMasterManager.clusterHasActiveMaster.get()); activeMasterManager.blockUntilBecomingActiveMaster(100, status); assertTrue(activeMasterManager.clusterHasActiveMaster.get()); assertMaster(zk, master); }
boolean blockUntilBecomingActiveMaster( int checkInterval, MonitoredTask startupStatus) { String backupZNode = ZKUtil.joinZNode( this.watcher.backupMasterAddressesZNode, this.sn.toString()); while (!(master.isAborted() || master.isStopped())) { startupStatus.setStatus("Trying to register in ZK as active master"); try { if (MasterAddressTracker.setMasterAddress(this.watcher, this.watcher.getMasterAddressZNode(), this.sn)) { if (ZKUtil.checkExists(this.watcher, backupZNode) != -1) { LOG.info("Deleting ZNode for " + backupZNode + " from backup master directory"); ZKUtil.deleteNodeFailSilent(this.watcher, backupZNode); } ZNodeClearer.writeMyEphemeralNodeOnDisk(this.sn.toString()); startupStatus.setStatus("Successfully registered as active master."); this.clusterHasActiveMaster.set(true); LOG.info("Registered Active Master=" + this.sn); return true; } this.clusterHasActiveMaster.set(true); String msg; byte[] bytes = ZKUtil.getDataAndWatch(this.watcher, this.watcher.getMasterAddressZNode()); 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 { ServerName currentMaster; try { currentMaster = ServerName.parseFrom(bytes); } catch (DeserializationException e) { LOG.warn("Failed parse", e); continue; } if (ServerName.isSameHostnameAndPort(currentMaster, this.sn)) { msg = ("Current master has this master's address, " + currentMaster + "; master was restarted? Deleting node."); ZKUtil.deleteNode(this.watcher, this.watcher.getMasterAddressZNode()); ZNodeClearer.deleteMyEphemeralNodeOnDisk(); } else { msg = "Another master is the active master, " + currentMaster + "; waiting to become the next active master"; } } LOG.info(msg); startupStatus.setStatus(msg); } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (clusterHasActiveMaster.get() && !master.isStopped()) { try { clusterHasActiveMaster.wait(checkInterval); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (clusterShutDown.get()) { this.master.stop( "Cluster went down before this master became active"); } } } return false; }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster( int checkInterval, MonitoredTask startupStatus) { String backupZNode = ZKUtil.joinZNode( this.watcher.backupMasterAddressesZNode, this.sn.toString()); while (!(master.isAborted() || master.isStopped())) { startupStatus.setStatus("Trying to register in ZK as active master"); try { if (MasterAddressTracker.setMasterAddress(this.watcher, this.watcher.getMasterAddressZNode(), this.sn)) { if (ZKUtil.checkExists(this.watcher, backupZNode) != -1) { LOG.info("Deleting ZNode for " + backupZNode + " from backup master directory"); ZKUtil.deleteNodeFailSilent(this.watcher, backupZNode); } ZNodeClearer.writeMyEphemeralNodeOnDisk(this.sn.toString()); startupStatus.setStatus("Successfully registered as active master."); this.clusterHasActiveMaster.set(true); LOG.info("Registered Active Master=" + this.sn); return true; } this.clusterHasActiveMaster.set(true); String msg; byte[] bytes = ZKUtil.getDataAndWatch(this.watcher, this.watcher.getMasterAddressZNode()); 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 { ServerName currentMaster; try { currentMaster = ServerName.parseFrom(bytes); } catch (DeserializationException e) { LOG.warn("Failed parse", e); continue; } if (ServerName.isSameHostnameAndPort(currentMaster, this.sn)) { msg = ("Current master has this master's address, " + currentMaster + "; master was restarted? Deleting node."); ZKUtil.deleteNode(this.watcher, this.watcher.getMasterAddressZNode()); ZNodeClearer.deleteMyEphemeralNodeOnDisk(); } else { msg = "Another master is the active master, " + currentMaster + "; waiting to become the next active master"; } } LOG.info(msg); startupStatus.setStatus(msg); } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (clusterHasActiveMaster.get() && !master.isStopped()) { try { clusterHasActiveMaster.wait(checkInterval); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (clusterShutDown.get()) { this.master.stop( "Cluster went down before this master became active"); } } } return false; } }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster( int checkInterval, MonitoredTask startupStatus) { String backupZNode = ZKUtil.joinZNode( this.watcher.backupMasterAddressesZNode, this.sn.toString()); while (!(master.isAborted() || master.isStopped())) { startupStatus.setStatus("Trying to register in ZK as active master"); try { if (MasterAddressTracker.setMasterAddress(this.watcher, this.watcher.getMasterAddressZNode(), this.sn)) { if (ZKUtil.checkExists(this.watcher, backupZNode) != -1) { LOG.info("Deleting ZNode for " + backupZNode + " from backup master directory"); ZKUtil.deleteNodeFailSilent(this.watcher, backupZNode); } ZNodeClearer.writeMyEphemeralNodeOnDisk(this.sn.toString()); startupStatus.setStatus("Successfully registered as active master."); this.clusterHasActiveMaster.set(true); LOG.info("Registered Active Master=" + this.sn); return true; } this.clusterHasActiveMaster.set(true); String msg; byte[] bytes = ZKUtil.getDataAndWatch(this.watcher, this.watcher.getMasterAddressZNode()); 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 { ServerName currentMaster; try { currentMaster = ServerName.parseFrom(bytes); } catch (DeserializationException e) { LOG.warn("Failed parse", e); continue; } if (ServerName.isSameHostnameAndPort(currentMaster, this.sn)) { msg = ("Current master has this master's address, " + currentMaster + "; master was restarted? Deleting node."); ZKUtil.deleteNode(this.watcher, this.watcher.getMasterAddressZNode()); ZNodeClearer.deleteMyEphemeralNodeOnDisk(); } else { msg = "Another master is the active master, " + currentMaster + "; waiting to become the next active master"; } } LOG.info(msg); startupStatus.setStatus(msg); } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (clusterHasActiveMaster.get() && !master.isStopped()) { try { clusterHasActiveMaster.wait(checkInterval); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (clusterShutDown.get()) { this.master.stop( "Cluster went down before this master became active"); } } } return false; } ActiveMasterManager(ZooKeeperWatcher watcher, ServerName sn, Server master); }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster( int checkInterval, MonitoredTask startupStatus) { String backupZNode = ZKUtil.joinZNode( this.watcher.backupMasterAddressesZNode, this.sn.toString()); while (!(master.isAborted() || master.isStopped())) { startupStatus.setStatus("Trying to register in ZK as active master"); try { if (MasterAddressTracker.setMasterAddress(this.watcher, this.watcher.getMasterAddressZNode(), this.sn)) { if (ZKUtil.checkExists(this.watcher, backupZNode) != -1) { LOG.info("Deleting ZNode for " + backupZNode + " from backup master directory"); ZKUtil.deleteNodeFailSilent(this.watcher, backupZNode); } ZNodeClearer.writeMyEphemeralNodeOnDisk(this.sn.toString()); startupStatus.setStatus("Successfully registered as active master."); this.clusterHasActiveMaster.set(true); LOG.info("Registered Active Master=" + this.sn); return true; } this.clusterHasActiveMaster.set(true); String msg; byte[] bytes = ZKUtil.getDataAndWatch(this.watcher, this.watcher.getMasterAddressZNode()); 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 { ServerName currentMaster; try { currentMaster = ServerName.parseFrom(bytes); } catch (DeserializationException e) { LOG.warn("Failed parse", e); continue; } if (ServerName.isSameHostnameAndPort(currentMaster, this.sn)) { msg = ("Current master has this master's address, " + currentMaster + "; master was restarted? Deleting node."); ZKUtil.deleteNode(this.watcher, this.watcher.getMasterAddressZNode()); ZNodeClearer.deleteMyEphemeralNodeOnDisk(); } else { msg = "Another master is the active master, " + currentMaster + "; waiting to become the next active master"; } } LOG.info(msg); startupStatus.setStatus(msg); } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (clusterHasActiveMaster.get() && !master.isStopped()) { try { clusterHasActiveMaster.wait(checkInterval); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (clusterShutDown.get()) { this.master.stop( "Cluster went down before this master became active"); } } } return false; } ActiveMasterManager(ZooKeeperWatcher watcher, ServerName sn, Server master); @Override void nodeCreated(String path); @Override void nodeDeleted(String path); void stop(); }
ActiveMasterManager extends ZooKeeperListener { boolean blockUntilBecomingActiveMaster( int checkInterval, MonitoredTask startupStatus) { String backupZNode = ZKUtil.joinZNode( this.watcher.backupMasterAddressesZNode, this.sn.toString()); while (!(master.isAborted() || master.isStopped())) { startupStatus.setStatus("Trying to register in ZK as active master"); try { if (MasterAddressTracker.setMasterAddress(this.watcher, this.watcher.getMasterAddressZNode(), this.sn)) { if (ZKUtil.checkExists(this.watcher, backupZNode) != -1) { LOG.info("Deleting ZNode for " + backupZNode + " from backup master directory"); ZKUtil.deleteNodeFailSilent(this.watcher, backupZNode); } ZNodeClearer.writeMyEphemeralNodeOnDisk(this.sn.toString()); startupStatus.setStatus("Successfully registered as active master."); this.clusterHasActiveMaster.set(true); LOG.info("Registered Active Master=" + this.sn); return true; } this.clusterHasActiveMaster.set(true); String msg; byte[] bytes = ZKUtil.getDataAndWatch(this.watcher, this.watcher.getMasterAddressZNode()); 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 { ServerName currentMaster; try { currentMaster = ServerName.parseFrom(bytes); } catch (DeserializationException e) { LOG.warn("Failed parse", e); continue; } if (ServerName.isSameHostnameAndPort(currentMaster, this.sn)) { msg = ("Current master has this master's address, " + currentMaster + "; master was restarted? Deleting node."); ZKUtil.deleteNode(this.watcher, this.watcher.getMasterAddressZNode()); ZNodeClearer.deleteMyEphemeralNodeOnDisk(); } else { msg = "Another master is the active master, " + currentMaster + "; waiting to become the next active master"; } } LOG.info(msg); startupStatus.setStatus(msg); } catch (KeeperException ke) { master.abort("Received an unexpected KeeperException, aborting", ke); return false; } synchronized (this.clusterHasActiveMaster) { while (clusterHasActiveMaster.get() && !master.isStopped()) { try { clusterHasActiveMaster.wait(checkInterval); } catch (InterruptedException e) { LOG.debug("Interrupted waiting for master to die", e); } } if (clusterShutDown.get()) { this.master.stop( "Cluster went down before this master became active"); } } } return false; } ActiveMasterManager(ZooKeeperWatcher watcher, ServerName sn, Server master); @Override void nodeCreated(String path); @Override void nodeDeleted(String path); void stop();  }