import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Verifies that the specified region is in the specified state in ZooKeeper.
 * <p>
 * Returns true if region is in transition and in the specified state in
 * ZooKeeper.  Returns false if the region does not exist in ZK or is in
 * a different state.
 * <p>
 * Method synchronizes() with ZK so will yield an up-to-date result but is
 * a slow read.
 * @param zkw
 * @param region
 * @param expectedState
 * @return true if region exists and is in expected state
 */
public static boolean verifyRegionState(ZooKeeperWatcher zkw,
    HRegionInfo region, EventType expectedState)
throws KeeperException {
  String encoded = region.getEncodedName();

  String node = getNodeName(zkw, encoded);
  zkw.sync(node);

  // Read existing data of the node
  byte [] existingBytes = null;
  try {
    existingBytes = ZKUtil.getDataAndWatch(zkw, node);
  } catch (KeeperException.NoNodeException nne) {
    return false;
  } catch (KeeperException e) {
    throw e;
  }
  if (existingBytes == null) return false;
  RegionTransitionData existingData =
    RegionTransitionData.fromBytes(existingBytes);
  if (existingData.getEventType() == expectedState){
    return true;
  }
  return false;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * New unassigned node has been created.
 *
 * <p>This happens when an RS begins the OPENING, SPLITTING or CLOSING of a
 * region by creating a znode.
 *
 * <p>When this happens we must:
 * <ol>
 *   <li>Watch the node for further children changed events</li>
 *   <li>Watch all new children for changed events</li>
 * </ol>
 */
@Override
public void nodeChildrenChanged(String path) {
  if(path.equals(watcher.assignmentZNode)) {
    try {
      List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(watcher,
          watcher.assignmentZNode);
      if (children != null) {
        Stat stat = new Stat();
        for (String child : children) {
          stat.setVersion(0);
          RegionTransitionData data = ZKAssign.getDataAndWatch(watcher,
              ZKUtil.joinZNode(watcher.assignmentZNode, child), stat);
          // See HBASE-7551, handle splitting here as well, in case we miss the node change event
          if (stat.getVersion() > 0 && data.getEventType() == EventType.RS_ZK_REGION_SPLITTING) {
            handleRegion(data, stat.getVersion());
          }
        }
      }
    } catch(KeeperException e) {
      master.abort("Unexpected ZK exception reading unassigned children", e);
    }
  }
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * @param path
 * @return True if znode is in SPLIT or SPLITTING state.
 * @throws KeeperException Can happen if the znode went away in meantime.
 */
private boolean isSplitOrSplitting(final String path) throws KeeperException {
  boolean result = false;
  // This may fail if the SPLIT or SPLITTING znode gets cleaned up before we
  // can get data from it.
  RegionTransitionData data = ZKAssign.getData(master.getZooKeeper(), path);
  EventType evt = data.getEventType();
  switch (evt) {
  case RS_ZK_REGION_SPLIT:
  case RS_ZK_REGION_SPLITTING:
    result = true;
    break;
  default:
    break;
  }
  return result;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Transition ZK node from OFFLINE to OPENING. The master will get a callback and will know that
 * the region is now ours.
 * @param hri HRegionInfo whose znode we are updating
 * @param versionOfOfflineNode Version Of OfflineNode that needs to be compared before changing
 *          the node's state from OFFLINE
 * @throws IOException
 */
int transitionZookeeperOfflineToOpening(final HRegionInfo hri, int versionOfOfflineNode)
    throws IOException {
  // TODO: should also handle transition from CLOSED?
  int version = -1;
  try {
    // Initialize the znode version.
    version =
        ZKAssign.transitionNode(this.zooKeeper, hri, this.getServerName(),
          EventType.M_ZK_REGION_OFFLINE, EventType.RS_ZK_REGION_OPENING, versionOfOfflineNode);
  } catch (KeeperException e) {
    LOG.error("Error transition from OFFLINE to OPENING for region=" + hri.getEncodedName(), e);
  }
  if (version == -1) {
    // TODO: Fix this sloppyness. The exception should be coming off zk
    // directly, not an
    // intepretation at this high-level (-1 when we call transitionNode can
    // mean many things).
    throw new IOException("Failed transition from OFFLINE to OPENING for region="
        + hri.getEncodedName());
  }
  return version;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
@Override
public void readFields(DataInput in) throws IOException {
  // the event type byte
  eventType = EventType.values()[in.readShort()];
  // the timestamp
  stamp = in.readLong();
  // the encoded name of the region being transitioned
  regionName = Bytes.readByteArray(in);
  // remaining fields are optional so prefixed with boolean
  // the name of the regionserver sending the data
  if (in.readBoolean()) {
    byte [] versionedBytes = Bytes.readByteArray(in);
    this.origin = ServerName.parseVersionedServerName(versionedBytes);
  }
  if (in.readBoolean()) {
    this.payload = Bytes.readByteArray(in);
  }
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Modify table is async so wait on completion of the table operation in master.
 * @param tableName
 * @param htd
 * @throws IOException
 */
private void modifyTable(final byte [] tableName, final HTableDescriptor htd)
throws IOException {
  MasterServices services = TEST_UTIL.getMiniHBaseCluster().getMaster();
  ExecutorService executor = services.getExecutorService();
  AtomicBoolean done = new AtomicBoolean(false);
  executor.registerListener(EventType.C_M_MODIFY_TABLE, new DoneListener(done));
  this.admin.modifyTable(tableName, htd);
  while (!done.get()) {
    synchronized (done) {
      try {
        done.wait(100);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
  executor.unregisterListener(EventType.C_M_MODIFY_TABLE);
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
@Test (timeout=300000) public void testCloseRegion()
throws Exception {
  LOG.info("Running testCloseRegion");
  MiniHBaseCluster cluster = TEST_UTIL.getHBaseCluster();
  LOG.info("Number of region servers = " + cluster.getLiveRegionServerThreads().size());

  int rsIdx = 0;
  HRegionServer regionServer = TEST_UTIL.getHBaseCluster().getRegionServer(rsIdx);
  HRegionInfo hri = getNonMetaRegion(regionServer.getOnlineRegions());
  LOG.debug("Asking RS to close region " + hri.getRegionNameAsString());

  AtomicBoolean closeEventProcessed = new AtomicBoolean(false);
  EventHandlerListener listener =
    new CloseRegionEventListener(hri.getRegionNameAsString(),
        closeEventProcessed);
  cluster.getMaster().executorService.registerListener(EventType.RS_ZK_REGION_CLOSED, listener);

  cluster.getMaster().assignmentManager.unassign(hri);

  while (!closeEventProcessed.get()) {
    Threads.sleep(100);
  }
  LOG.info("Done with testCloseRegion");
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Creates a new ephemeral node in the SPLITTING state for the specified region.
 * Create it ephemeral in case regionserver dies mid-split.
 *
 * <p>Does not transition nodes from other states.  If a node already exists
 * for this region, a {@link NodeExistsException} will be thrown.
 *
 * @param zkw zk reference
 * @param region region to be created as offline
 * @param serverName server event originates from
 * @return Version of znode created.
 * @throws KeeperException
 * @throws IOException
 */
// Copied from SplitTransaction rather than open the method over there in
// the regionserver package.
private static int createNodeSplitting(final ZooKeeperWatcher zkw,
    final HRegionInfo region, final ServerName serverName)
throws KeeperException, IOException {
  RegionTransitionData data =
    new RegionTransitionData(EventType.RS_ZK_REGION_SPLITTING,
      region.getRegionName(), serverName);

  String node = ZKAssign.getNodeName(zkw, region.getEncodedName());
  if (!ZKUtil.createEphemeralNodeAndWatch(zkw, node, data.getBytes())) {
    throw new IOException("Failed create of ephemeral " + node);
  }
  // Transition node from SPLITTING to SPLITTING and pick up version so we
  // can be sure this znode is ours; version is needed deleting.
  return transitionNodeSplitting(zkw, region, serverName, -1);
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
private void OpenRegion(Server server, RegionServerServices rss,
      HTableDescriptor htd, HRegionInfo hri)
      throws IOException, NodeExistsException, KeeperException {
      // Create it OFFLINE node, which is what Master set before sending OPEN RPC


  ZKAssign.createNodeOffline(server.getZooKeeper(), hri, server.getServerName());
  int version = ZKAssign.transitionNodeOpening(server.getZooKeeper(), hri, server.getServerName());
  OpenRegionHandler openHandler = new OpenRegionHandler(server, rss, hri, htd, version);
  openHandler.process();
  RegionTransitionData data = ZKAssign.getData(server.getZooKeeper(), hri.getEncodedName());

  // delete the node, which is what Master do after the region is opened
  ZKAssign.deleteNode(server.getZooKeeper(), hri.getEncodedName(),
    EventType.RS_ZK_REGION_OPENED);
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
@Test
public void testFailedOpenRegion() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = new MockRegionServerServices();

  // Create it OFFLINE, which is what it expects
  ZKAssign.createNodeOffline(server.getZooKeeper(), TEST_HRI, server.getServerName());
  ZKAssign.transitionNodeOpening(server.getZooKeeper(), TEST_HRI, server.getServerName());

  // Create the handler
  OpenRegionHandler handler =
    new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD) {
      @Override
      HRegion openRegion() {
        // Fake failure of opening a region due to an IOE, which is caught
        return null;
      }
  };
  handler.process();

  // Handler should have transitioned it to FAILED_OPEN
  RegionTransitionData data =
    ZKAssign.getData(server.getZooKeeper(), TEST_HRI.getEncodedName());
  assertEquals(EventType.RS_ZK_REGION_FAILED_OPEN, data.getEventType());
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
@Test
public void testFailedUpdateMeta() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = new MockRegionServerServices();

  // Create it OFFLINE, which is what it expects
  ZKAssign.createNodeOffline(server.getZooKeeper(), TEST_HRI, server.getServerName());
  ZKAssign.transitionNodeOpening(server.getZooKeeper(), TEST_HRI, server.getServerName());
  // Create the handler
  OpenRegionHandler handler =
    new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD) {
      @Override
      boolean updateMeta(final HRegion r) {
        // Fake failure of updating META
        return false;
      }
  };
  handler.process();

  // Handler should have transitioned it to FAILED_OPEN
  RegionTransitionData data =
    ZKAssign.getData(server.getZooKeeper(), TEST_HRI.getEncodedName());
  assertEquals(EventType.RS_ZK_REGION_FAILED_OPEN, data.getEventType());
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Transition ZK node from OFFLINE to OPENING. The master will get a callback
 * and will know that the region is now ours.
 *
 * @param hri
 *          HRegionInfo whose znode we are updating
 * @param versionOfOfflineNode
 *          Version Of OfflineNode that needs to be compared before changing
 *          the node's state from OFFLINE
 * @throws IOException
 */
int transitionZookeeperOfflineToOpening(final HRegionInfo hri, int versionOfOfflineNode)
    throws IOException {
  // TODO: should also handle transition from CLOSED?
  int version = -1;
  try {
    // Initialize the znode version.
    version = ZKAssign.transitionNode(this.zooKeeper, hri, this.getServerName(),
        EventType.M_ZK_REGION_OFFLINE, EventType.RS_ZK_REGION_OPENING, versionOfOfflineNode);
  } catch (KeeperException e) {
    LOG.error("Error transition from OFFLINE to OPENING for region=" + hri.getEncodedName(), e);
  }
  if (version == -1) {
    // TODO: Fix this sloppyness. The exception should be coming off zk
    // directly, not an
    // intepretation at this high-level (-1 when we call transitionNode can
    // mean many things).
    throw new IOException("Failed transition from OFFLINE to OPENING for region="
        + hri.getEncodedName());
  }
  return version;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Creates a new ephemeral node in the SPLITTING state for the specified region.
 * Create it ephemeral in case regionserver dies mid-split.
 *
 * <p>Does not transition nodes from other states.  If a node already exists
 * for this region, a {@link NodeExistsException} will be thrown.
 *
 * @param zkw zk reference
 * @param region region to be created as offline
 * @param serverName server event originates from
 * @return Version of znode created.
 * @throws KeeperException 
 * @throws IOException 
 */
private static int createNodeSplitting(final ZooKeeperWatcher zkw,
    final HRegionInfo region, final ServerName serverName)
throws KeeperException, IOException {
  LOG.debug(zkw.prefix("Creating ephemeral node for " +
    region.getEncodedName() + " in SPLITTING state"));
  RegionTransitionData data =
    new RegionTransitionData(EventType.RS_ZK_REGION_SPLITTING,
      region.getRegionName(), serverName);

  String node = ZKAssign.getNodeName(zkw, region.getEncodedName());
  if (!ZKUtil.createEphemeralNodeAndWatch(zkw, node, data.getBytes())) {
    throw new IOException("Failed create of ephemeral " + node);
  }
  // Transition node from SPLITTING to SPLITTING and pick up version so we
  // can be sure this znode is ours; version is needed deleting.
  return transitionNodeSplitting(zkw, region, serverName, -1);
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
public static void createNodeOffline(ZooKeeperWatcher zkw, HRegionInfo region,
    ServerName serverName, final EventType event)
throws KeeperException, KeeperException.NodeExistsException {
  LOG.debug(zkw.prefix("Creating unassigned node for " +
    region.getEncodedName() + " in OFFLINE state"));
  RegionTransitionData data = new RegionTransitionData(event,
    region.getRegionName(), serverName);
  String node = getNodeName(zkw, region.getEncodedName());
  ZKUtil.createAndWatch(zkw, node, data.getBytes());
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
/**
 * Deletes an existing unassigned node that is in the specified state for the
 * specified region.
 *
 * <p>If a node does not already exist for this region, a
 * {@link NoNodeException} will be thrown.
 *
 * <p>No watcher is set whether this succeeds or not.
 *
 * <p>Returns false if the node was not in the proper state but did exist.
 *
 * <p>This method is used when a region finishes opening/closing.
 * The Master acknowledges completion
 * of the specified regions transition to being closed/opened.
 *
 * @param zkw zk reference
 * @param regionName region to be deleted from zk
 * @param expectedState state region must be in for delete to complete
 * @param expectedVersion of the znode that is to be deleted.
 *        If expectedVersion need not be compared while deleting the znode
 *        pass -1
 * @throws KeeperException if unexpected zookeeper exception
 * @throws KeeperException.NoNodeException if node does not exist
 */
public static boolean deleteNode(ZooKeeperWatcher zkw, String regionName,
    EventType expectedState, int expectedVersion)
throws KeeperException, KeeperException.NoNodeException {
  LOG.debug(zkw.prefix("Deleting existing unassigned " +
    "node for " + regionName + " that is in expected state " + expectedState));
  String node = getNodeName(zkw, regionName);
  zkw.sync(node);
  Stat stat = new Stat();
  byte [] bytes = ZKUtil.getDataNoWatch(zkw, node, stat);
  if (bytes == null) {
    // If it came back null, node does not exist.
    throw KeeperException.create(Code.NONODE);
  }
  RegionTransitionData data = RegionTransitionData.fromBytes(bytes);
  if (!data.getEventType().equals(expectedState)) {
    LOG.warn(zkw.prefix("Attempting to delete unassigned " +
      "node " + regionName + " in " + expectedState +
      " state but node is in " + data.getEventType() + " state"));
    return false;
  }
  if (expectedVersion != -1
      && stat.getVersion() != expectedVersion) {
    LOG.warn("The node " + regionName + " we are trying to delete is not" +
      " the expected one. Got a version mismatch");
    return false;
  }
  if(!ZKUtil.deleteNode(zkw, node, stat.getVersion())) {
    LOG.warn(zkw.prefix("Attempting to delete " +
        "unassigned node " + regionName + " in " + expectedState +
        " state but after verifying state, we got a version mismatch"));
    return false;
  }
  LOG.debug(zkw.prefix("Successfully deleted unassigned node for region " +
      regionName + " in expected state " + expectedState));
  return true;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
private void processOpeningState(HRegionInfo regionInfo) {
  LOG.info("Region has been OPENING for too " + "long, reassigning region="
      + regionInfo.getRegionNameAsString());
  // Should have a ZK node in OPENING state
  try {
    String node = ZKAssign.getNodeName(watcher, regionInfo.getEncodedName());
    Stat stat = new Stat();
    RegionTransitionData dataInZNode = ZKAssign.getDataNoWatch(watcher, node,
        stat);
    if (dataInZNode == null) {
      LOG.warn("Data is null, node " + node + " no longer exists");
      return;
    }
    if (dataInZNode.getEventType() == EventType.RS_ZK_REGION_OPENED) {
      LOG.debug("Region has transitioned to OPENED, allowing "
          + "watched event handlers to process");
      return;
    } else if (dataInZNode.getEventType() != EventType.RS_ZK_REGION_OPENING &&
        dataInZNode.getEventType() != EventType.RS_ZK_REGION_FAILED_OPEN ) {
      LOG.warn("While timing out a region in state OPENING, "
          + "found ZK node in unexpected state: "
          + dataInZNode.getEventType());
      return;
    }
    invokeAssign(regionInfo);
  } catch (KeeperException ke) {
    LOG.error("Unexpected ZK exception timing out CLOSING region", ke);
    return;
  }
  return;
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
private static void cleanZK(final Server server, final HRegionInfo hri) {
  try {
    // Only delete if its in expected state; could have been hijacked.
    ZKAssign.deleteNode(server.getZooKeeper(), hri.getEncodedName(),
      EventType.RS_ZK_REGION_SPLITTING);
  } catch (KeeperException e) {
    server.abort("Failed cleanup of " + hri.getRegionNameAsString(), e);
  }
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
private static int tickleNodeSplit(ZooKeeperWatcher zkw,
    HRegionInfo parent, HRegionInfo a, HRegionInfo b, ServerName serverName,
    final int znodeVersion)
throws KeeperException, IOException {
  byte [] payload = Writables.getBytes(a, b);
  return ZKAssign.transitionNode(zkw, parent, serverName,
    EventType.RS_ZK_REGION_SPLIT, EventType.RS_ZK_REGION_SPLIT,
    znodeVersion, payload);
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
protected Executor(String name, int maxThreads,
    final Map<EventHandler.EventType, EventHandlerListener> eventHandlerListeners) {
  this.id = seqids.incrementAndGet();
  this.name = name;
  this.eventHandlerListeners = eventHandlerListeners;
  // create the thread pool executor
  this.threadPoolExecutor = new TrackingThreadPoolExecutor(
      maxThreads, maxThreads,
      keepAliveTimeInMillis, TimeUnit.MILLISECONDS, q);
  // name the threads for this threadpool
  ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
  tfb.setNameFormat(this.name + "-%d");
  this.threadPoolExecutor.setThreadFactory(tfb.build());
}
 

import org.apache.hadoop.hbase.executor.EventHandler.EventType; //导入依赖的package包/类
@Test (timeout=300000) public void testRestartClusterAfterKill()
throws Exception {
  UTIL.startMiniZKCluster();
  ZooKeeperWatcher zooKeeper =
    new ZooKeeperWatcher(UTIL.getConfiguration(), "cluster1", null, true);

  // create the unassigned region, throw up a region opened state for META
  String unassignedZNode = zooKeeper.assignmentZNode;
  ZKUtil.createAndFailSilent(zooKeeper, unassignedZNode);

  ServerName sn = new ServerName(HMaster.MASTER, -1, System.currentTimeMillis());

  ZKAssign.createNodeOffline(zooKeeper, HRegionInfo.ROOT_REGIONINFO, sn);

  ZKAssign.createNodeOffline(zooKeeper, HRegionInfo.FIRST_META_REGIONINFO, sn);

  LOG.debug("Created UNASSIGNED zNode for ROOT and META regions in state " +
      EventType.M_ZK_REGION_OFFLINE);

  // start the HB cluster
  LOG.info("Starting HBase cluster...");
  UTIL.startMiniCluster(2);

  UTIL.createTable(TABLENAME, FAMILIES);
  LOG.info("Created a table, waiting for table to be available...");
  UTIL.waitTableAvailable(TABLENAME, 60*1000);

  LOG.info("Master deleted unassigned region and started up successfully.");
}