HBASE-1448 Add a node in ZK to tell all masters to shutdown

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@781134 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -302,6 +302,7 @@ Release 0.20.0 - Unreleased
    HBASE-1445  Add the ability to start a master from any machine
    HBASE-1474  Add zk attributes to list of attributes 
                in master and regionserver UIs
+   HBASE-1448  Add a node in ZK to tell all masters to shutdown
 
   OPTIMIZATIONS
    HBASE-1412  Change values for delete column and column family in KeyValue

src/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -243,7 +243,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
     this.sleeper = new Sleeper(this.threadWakeFrequency, this.closed);
     
     zooKeeperWrapper = new ZooKeeperWrapper(conf);
-    zkMasterAddressWatcher = new ZKMasterAddressWatcher(zooKeeperWrapper);
+    zkMasterAddressWatcher = new ZKMasterAddressWatcher(this);
     serverManager = new ServerManager(this);
     regionManager = new RegionManager(this);
 
@@ -257,7 +257,12 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
   private void writeAddressToZooKeeper() {
     while (true) {
       zkMasterAddressWatcher.waitForMasterAddressAvailability();
-      if (zooKeeperWrapper.writeMasterAddress(address)) {
+      // Check if we need to shutdown instead of taking control
+      if(this.shutdownRequested.get())
+      {
+        return;
+      } else if(zooKeeperWrapper.writeMasterAddress(address)) {
+        zooKeeperWrapper.setClusterState(true);
         return;
       }
     }
@@ -666,6 +671,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
   public void shutdown() {
     LOG.info("Cluster shutdown requested. Starting to quiesce servers");
     this.shutdownRequested.set(true);
+    this.zooKeeperWrapper.setClusterState(false);
   }
 
   public void createTable(HTableDescriptor desc)
@@ -1088,6 +1094,10 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
             Constructor<? extends HMaster> c =
               masterClass.getConstructor(HBaseConfiguration.class);
             HMaster master = c.newInstance(conf);
+            if(master.shutdownRequested.get()) {
+              LOG.info("Won't bring the Master up as a shutdown is requested");
+              return;
+            }
             master.start();
           }
         } catch (Throwable t) {

src/java/org/apache/hadoop/hbase/master/ZKMasterAddressWatcher.java

@@ -38,13 +38,15 @@ public class ZKMasterAddressWatcher implements Watcher {
   private static final Log LOG = LogFactory.getLog(ZKMasterAddressWatcher.class);
 
   private final ZooKeeperWrapper zooKeeper;
+  private final HMaster master;
 
   /**
    * Create a watcher with a ZooKeeperWrapper instance.
    * @param zooKeeper ZooKeeperWrapper to use to talk to ZooKeeper.
    */
-  public ZKMasterAddressWatcher(ZooKeeperWrapper zooKeeper) {
-    this.zooKeeper = zooKeeper;
+  public ZKMasterAddressWatcher(HMaster master) {
+    this.master = master;
+    this.zooKeeper = master.getZooKeeperWrapper();
   }
 
   /**
@@ -53,9 +55,22 @@ public class ZKMasterAddressWatcher implements Watcher {
   @Override
   public synchronized void process(WatchedEvent event) {
     EventType type = event.getType();
+    LOG.debug(("Got event " + type + " with path " + event.getPath()));
     if (type.equals(EventType.NodeDeleted)) {
-      LOG.debug("Master address ZNode deleted, notifying waiting masters");
-      notifyAll();
+      if(event.getPath().equals(this.zooKeeper.clusterStateZNode)) {
+        LOG.info("The cluster was shutdown while waiting, shutting down" +
+            " this master.");
+        master.shutdownRequested.set(true);
+      }
+      else {
+        LOG.debug("Master address ZNode deleted, notifying waiting masters");
+        notifyAll();
+      }
+    }
+    else if(type.equals(EventType.NodeCreated) && 
+        event.getPath().equals(this.zooKeeper.clusterStateZNode)) {
+      LOG.debug("Resetting the watch on the cluster state node.");
+      this.zooKeeper.setClusterStateWatch(this);
     }
   }
 
@@ -66,7 +81,9 @@ public class ZKMasterAddressWatcher implements Watcher {
   public synchronized void waitForMasterAddressAvailability() {
     while (zooKeeper.readMasterAddress(this) != null) {
       try {
-        LOG.debug("Waiting for master address ZNode to be deleted");
+        LOG.debug("Waiting for master address ZNode to be deleted " +
+            "and watching the cluster state node");
+        this.zooKeeper.setClusterStateWatch(this);
         wait();
       } catch (InterruptedException e) {
       }

src/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWrapper.java

@@ -64,10 +64,11 @@ public class ZooKeeperWrapper implements HConstants {
   private final ZooKeeper zooKeeper;
   private final WatcherWrapper watcher;
 
-  private final String rootRegionZNode;
-  private final String outOfSafeModeZNode;
-  private final String rsZNode;
-  private final String masterElectionZNode;
+  public final String rootRegionZNode;
+  public final String outOfSafeModeZNode;
+  public final String rsZNode;
+  public final String masterElectionZNode;
+  public final String clusterStateZNode;
 
   /**
    * Create a ZooKeeperWrapper.
@@ -109,11 +110,14 @@ public class ZooKeeperWrapper implements HConstants {
     String rsZNodeName = conf.get("zookeeper.znode.rs", "rs");
     String masterAddressZNodeName = conf.get("zookeeper.znode.master",
       "master");
+    String stateZNodeName = conf.get("zookeeper.znode.state",
+    "shutdown");
     
     rootRegionZNode = getZNode(parentZNode, rootServerZNodeName);
     outOfSafeModeZNode = getZNode(parentZNode, outOfSafeModeZNodeName);
     rsZNode = getZNode(parentZNode, rsZNodeName);
     masterElectionZNode = getZNode(parentZNode, masterAddressZNodeName);
+    clusterStateZNode = getZNode(parentZNode, stateZNodeName);
   }
 
   /**
@@ -243,6 +247,49 @@ public class ZooKeeperWrapper implements HConstants {
   public HServerAddress readMasterAddress(Watcher watcher) {
     return readAddress(masterElectionZNode, watcher);
   }
+  
+  /**
+   * Watch the state of the cluster, up or down
+   * @param watcher Watcher to set on cluster state node
+   */
+  public void setClusterStateWatch(Watcher watcher) {
+    try {
+      zooKeeper.exists(clusterStateZNode, watcher);
+    } catch (InterruptedException e) {
+      LOG.warn("Failed to check on ZNode " + clusterStateZNode, e);
+    } catch (KeeperException e) {
+      LOG.warn("Failed to check on ZNode " + clusterStateZNode, e);
+    }
+  }
+  
+  /**
+   * Set the cluster state, up or down
+   * @param up True to write the node, false to delete it
+   * @return true if it worked, else it's false
+   */
+  public boolean setClusterState(boolean up) {
+    if (!ensureParentExists(clusterStateZNode)) {
+      return false;
+    }
+    try {
+      if(up) {
+        byte[] data = Bytes.toBytes("up");
+        zooKeeper.create(clusterStateZNode, data, 
+            Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        LOG.debug("State node wrote in ZooKeeper");
+      } else {
+        zooKeeper.delete(clusterStateZNode, -1);
+        LOG.debug("State node deleted in ZooKeeper");
+      }
+      return true;
+    } catch (InterruptedException e) {
+      LOG.warn("Failed to set state node in ZooKeeper", e);
+    } catch (KeeperException e) {
+      LOG.warn("Failed to set state node in ZooKeeper", e);
+    }
+
+    return false;
+  }
 
   /**
    * Set a watcher on the master address ZNode. The watcher will be set unless