Java 类org.apache.hadoop.hbase.executor.EventType 实例源码

public OpenedRegionHandler(Server server,
    AssignmentManager assignmentManager, HRegionInfo regionInfo,
    OpenRegionCoordination coordination,
    OpenRegionCoordination.OpenRegionDetails ord) {
  super(server, EventType.RS_ZK_REGION_OPENED);
  this.assignmentManager = assignmentManager;
  this.regionInfo = regionInfo;
  this.coordination = coordination;
  this.ord = ord;
  if(regionInfo.isMetaRegion()) {
    priority = OpenedPriority.META;
  } else if(regionInfo.getTable()
      .getNamespaceAsString().equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
    priority = OpenedPriority.SYSTEM;
  } else {
    priority = OpenedPriority.USER;
  }
}

public RestoreSnapshotHandler(final MasterServices masterServices,
    final SnapshotDescription snapshot, final HTableDescriptor htd) throws IOException {
  super(EventType.C_M_RESTORE_SNAPSHOT, htd.getTableName(), masterServices, masterServices);

  // Snapshot information
  this.snapshot = snapshot;

  // Monitor
  this.monitor = new ForeignExceptionDispatcher();

  // Check table exists.
  getTableDescriptor();

  // This is the new schema we are going to write out as this modification.
  this.hTableDescriptor = htd;

  this.status = TaskMonitor.get().createStatus(
    "Restoring  snapshot '" + snapshot.getName() + "' to table "
        + hTableDescriptor.getTableName());
}

private boolean deleteNodeInStates(String encodedName,
    String desc, ServerName sn, EventType... types) {
  try {
    for (EventType et: types) {
      if (ZKAssign.deleteNode(watcher, encodedName, et, sn)) {
        return true;
      }
    }
    LOG.info("Failed to delete the " + desc + " node for "
      + encodedName + ". The node type may not match");
  } catch (NoNodeException e) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("The " + desc + " node for " + encodedName + " already deleted");
    }
  } catch (KeeperException ke) {
    server.abort("Unexpected ZK exception deleting " + desc
      + " node for the region " + encodedName, ke);
  }
  return false;
}

/**
 * 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
 * @throws DeserializationException
 */
static boolean verifyRegionState(ZooKeeperWatcher zkw, HRegionInfo region, EventType expectedState)
throws KeeperException, DeserializationException {
  String encoded = region.getEncodedName();

  String node = ZKAssign.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;
  RegionTransition rt = RegionTransition.parseFrom(existingBytes);
  return rt.getEventType().equals(expectedState);
}

private void OpenRegion(Server server, RegionServerServices rss,
    HTableDescriptor htd, HRegionInfo hri, OpenRegionCoordination coordination)
throws IOException, NodeExistsException, KeeperException, DeserializationException {
  // Create it OFFLINE node, which is what Master set before sending OPEN RPC
  ZKAssign.createNodeOffline(server.getZooKeeper(), hri, server.getServerName());

  OpenRegionCoordination.OpenRegionDetails ord =
    coordination.getDetailsForNonCoordinatedOpening();
  OpenRegionHandler openHandler =
    new OpenRegionHandler(server, rss, hri, htd, -1, coordination, ord);
  rss.getRegionsInTransitionInRS().put(hri.getEncodedNameAsBytes(), Boolean.TRUE);
  openHandler.process();
  // This parse is not used?
  RegionTransition.parseFrom(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, server.getServerName());
}

/**
 * Fakes the regionserver-side zk transitions of a region open.
 * @param w ZooKeeperWatcher to use.
 * @param sn Name of the regionserver doing the 'opening'
 * @param hri Region we're 'opening'.
 * @throws KeeperException
 * @throws DeserializationException
 */
static void fakeRegionServerRegionOpenInZK(HMaster master,  final ZooKeeperWatcher w,
    final ServerName sn, final HRegionInfo hri)
  throws KeeperException, DeserializationException, InterruptedException {
  // Wait till the we region is ready to be open in RIT.
  waitForRegionPendingOpenInRIT(master.getAssignmentManager(), hri.getEncodedName());

  // Get current versionid else will fail on transition from OFFLINE to OPENING below
  int versionid = ZKAssign.getVersion(w, hri);
  assertNotSame(-1, versionid);
  // This uglyness below is what the openregionhandler on RS side does.  I
  // looked at exposing the method over in openregionhandler but its just a
  // one liner and its deep over in another package so just repeat it below.
  versionid = ZKAssign.transitionNode(w, hri, sn,
    EventType.M_ZK_REGION_OFFLINE, EventType.RS_ZK_REGION_OPENING, versionid);
  assertNotSame(-1, versionid);
  // Move znode from OPENING to OPENED as RS does on successful open.
  versionid = ZKAssign.transitionNodeOpened(w, hri, sn, versionid);
  assertNotSame(-1, versionid);
  // We should be done now.  The master open handler will notice the
  // transition and remove this regions znode.
}

protected OpenRegionHandler(final Server server,
    final RegionServerServices rsServices, final HRegionInfo regionInfo,
    final HTableDescriptor htd, EventType eventType,
    final int versionOfOfflineNode) {
  super(server, eventType);
  this.rsServices = rsServices;
  this.regionInfo = regionInfo;
  this.htd = htd;
  this.versionOfOfflineNode = versionOfOfflineNode;
  tomActivated = this.server.getConfiguration().
    getBoolean(AssignmentManager.ASSIGNMENT_TIMEOUT_MANAGEMENT,
      AssignmentManager.DEFAULT_ASSIGNMENT_TIMEOUT_MANAGEMENT);
  assignmentTimeout = this.server.getConfiguration().
    getInt(AssignmentManager.ASSIGNMENT_TIMEOUT,
      AssignmentManager.DEFAULT_ASSIGNMENT_TIMEOUT_DEFAULT);
}

/**
 * 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
 * @throws DeserializationException
 */
static boolean verifyRegionState(ZooKeeperWatcher zkw, HRegionInfo region, EventType expectedState)
throws KeeperException, DeserializationException {
  String encoded = region.getEncodedName();

  String node = ZKAssign.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;
  RegionTransition rt = RegionTransition.parseFrom(existingBytes);
  return rt.getEventType().equals(expectedState);
}

/**
 * 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 {
  RegionTransition rt =
    RegionTransition.createRegionTransition(EventType.RS_ZK_REGION_SPLITTING,
      region.getRegionName(), serverName);

  String node = ZKAssign.getNodeName(zkw, region.getEncodedName());
  if (!ZKUtil.createEphemeralNodeAndWatch(zkw, node, rt.toByteArray())) {
    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);
}

private void OpenRegion(Server server, RegionServerServices rss,
    HTableDescriptor htd, HRegionInfo hri, OpenRegionCoordination coordination)
throws IOException, NodeExistsException, KeeperException, DeserializationException {
  // Create it OFFLINE node, which is what Master set before sending OPEN RPC
  ZKAssign.createNodeOffline(server.getZooKeeper(), hri, server.getServerName());

  OpenRegionCoordination.OpenRegionDetails ord =
    coordination.getDetailsForNonCoordinatedOpening();
  OpenRegionHandler openHandler =
    new OpenRegionHandler(server, rss, hri, htd, coordination, ord);
  rss.getRegionsInTransitionInRS().put(hri.getEncodedNameAsBytes(), Boolean.TRUE);
  openHandler.process();
  // This parse is not used?
  RegionTransition.parseFrom(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, server.getServerName());
}

@Test
public void testTransitionToFailedOpenFromOffline() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = HTU.createMockRegionServerService(server.getServerName());
  // Create it OFFLINE, which is what it expects
  ZKAssign.createNodeOffline(server.getZooKeeper(), TEST_HRI, server.getServerName());
  // Create the handler
  OpenRegionHandler handler = new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD) {

    @Override
    boolean transitionZookeeperOfflineToOpening(String encodedName, int versionOfOfflineNode) {
      return false;
    }
  };
  rsServices.getRegionsInTransitionInRS().put(TEST_HRI.getEncodedNameAsBytes(), Boolean.TRUE);

  handler.process();

  RegionTransition rt = RegionTransition.parseFrom(ZKAssign.getData(server.getZooKeeper(),
      TEST_HRI.getEncodedName()));
  assertEquals(EventType.RS_ZK_REGION_FAILED_OPEN, rt.getEventType());
}

public OpenedRegionHandler(Server server,
    AssignmentManager assignmentManager, HRegionInfo regionInfo,
    ServerName sn, int expectedVersion) {
  super(server, EventType.RS_ZK_REGION_OPENED);
  this.assignmentManager = assignmentManager;
  this.regionInfo = regionInfo;
  this.sn = sn;
  this.expectedVersion = expectedVersion;
  if(regionInfo.isMetaRegion()) {
    priority = OpenedPriority.META;
  } else if(regionInfo.getTable()
      .getNamespaceAsString().equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
    priority = OpenedPriority.SYSTEM;
  } else {
    priority = OpenedPriority.USER;
  }
}

ServerShutdownHandler(final Server server, final MasterServices services,
    final DeadServer deadServers, final ServerName serverName, EventType type,
    final boolean shouldSplitHlog) {
  super(server, type);
  this.serverName = serverName;
  this.server = server;
  this.services = services;
  this.deadServers = deadServers;
  if (!this.deadServers.isDeadServer(this.serverName)) {
    LOG.warn(this.serverName + " is NOT in deadservers; it should be!");
  }
  this.shouldSplitHlog = shouldSplitHlog;
  this.distributedLogReplay = HLogSplitter.isDistributedLogReplay(server.getConfiguration());
  this.regionAssignmentWaitTimeout = server.getConfiguration().getInt(
    HConstants.LOG_REPLAY_WAIT_REGION_TIMEOUT, 15000);
}

/**
 * Transition ZK node from OFFLINE to OPENING.
 * @param regionInfo region info instance
 * @param ord - instance of open region details, for ZK implementation
 *   will include version Of OfflineNode that needs to be compared
 *   before changing the node's state from OFFLINE
 * @return True if successful transition.
 */
@Override
public boolean transitionFromOfflineToOpening(HRegionInfo regionInfo,
                                              OpenRegionDetails ord) {
  ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;

  // encoded name is used as znode encoded name in ZK
  final String encodedName = regionInfo.getEncodedName();

  // TODO: should also handle transition from CLOSED?
  try {
    // Initialize the znode version.
    zkOrd.setVersion(ZKAssign.transitionNode(watcher, regionInfo,
      zkOrd.getServerName(), EventType.M_ZK_REGION_OFFLINE,
      EventType.RS_ZK_REGION_OPENING, zkOrd.getVersionOfOfflineNode()));
  } catch (KeeperException e) {
    LOG.error("Error transition from OFFLINE to OPENING for region=" +
      encodedName, e);
    zkOrd.setVersion(-1);
    return false;
  }
  boolean b = isGoodVersion(zkOrd);
  if (!b) {
    LOG.warn("Failed transition from OFFLINE to OPENING for region=" +
      encodedName);
  }
  return b;
}

/**
 * Try to transition to open.
 *
 * This is not guaranteed to succeed, we just do our best.
 *
 * @param rsServices
 * @param hri Region we're working on.
 * @param ord Details about region open task
 * @return whether znode is successfully transitioned to FAILED_OPEN state.
 */
@Override
public boolean tryTransitionFromOfflineToFailedOpen(RegionServerServices rsServices,
                                                    final HRegionInfo hri,
                                                    OpenRegionDetails ord) {
  ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
  boolean result = false;
  final String name = hri.getRegionNameAsString();
  try {
    LOG.info("Opening of region " + hri + " failed, transitioning" +
      " from OFFLINE to FAILED_OPEN in ZK, expecting version " +
      zkOrd.getVersionOfOfflineNode());
    if (ZKAssign.transitionNode(
      rsServices.getZooKeeper(), hri,
      rsServices.getServerName(),
      EventType.M_ZK_REGION_OFFLINE,
      EventType.RS_ZK_REGION_FAILED_OPEN,
      zkOrd.getVersionOfOfflineNode()) == -1) {
      LOG.warn("Unable to mark region " + hri + " as FAILED_OPEN. " +
        "It's likely that the master already timed out this open " +
        "attempt, and thus another RS already has the region.");
    } else {
      result = true;
    }
  } catch (KeeperException e) {
    LOG.error("Failed transitioning node " + name + " from OFFLINE to FAILED_OPEN", e);
  }
  return result;
}

/**
 * This is not guaranteed to succeed, we just do our best.
 * @param hri Region we're working on.
 * @return whether znode is successfully transitioned to FAILED_OPEN state.
 */
@Override
public boolean tryTransitionFromOpeningToFailedOpen(final HRegionInfo hri,
                                                    OpenRegionDetails ord) {
  ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
  boolean result = false;
  final String name = hri.getRegionNameAsString();
  try {
    LOG.info("Opening of region " + hri + " failed, transitioning" +
      " from OPENING to FAILED_OPEN in ZK, expecting version " + zkOrd.getVersion());
    if (ZKAssign.transitionNode(
      watcher, hri,
      zkOrd.getServerName(),
      EventType.RS_ZK_REGION_OPENING,
      EventType.RS_ZK_REGION_FAILED_OPEN,
      zkOrd.getVersion()) == -1) {
      LOG.warn("Unable to mark region " + hri + " as FAILED_OPEN. " +
        "It's likely that the master already timed out this open " +
        "attempt, and thus another RS already has the region.");
    } else {
      result = true;
    }
  } catch (KeeperException e) {
    LOG.error("Failed transitioning node " + name +
      " from OPENING to FAILED_OPEN", e);
  }
  return result;
}

public DispatchMergingRegionHandler(final MasterServices services,
    final CatalogJanitor catalogJanitor, final HRegionInfo region_a,
    final HRegionInfo region_b, final boolean forcible) {
  super(services, EventType.C_M_MERGE_REGION);
  this.masterServices = services;
  this.catalogJanitor = catalogJanitor;
  this.region_a = region_a;
  this.region_b = region_b;
  this.forcible = forcible;
  this.timeout = server.getConfiguration().getInt(
      "hbase.master.regionmerge.timeout", 120 * 1000);
}

public CreateTableHandler(Server server, MasterFileSystem fileSystemManager,
    HTableDescriptor hTableDescriptor, Configuration conf, HRegionInfo [] newRegions,
    MasterServices masterServices) {
  super(server, EventType.C_M_CREATE_TABLE);

  this.fileSystemManager = fileSystemManager;
  this.hTableDescriptor = hTableDescriptor;
  this.conf = conf;
  this.newRegions = newRegions;
  this.assignmentManager = masterServices.getAssignmentManager();
  this.tableLockManager = masterServices.getTableLockManager();

  this.tableLock = this.tableLockManager.writeLock(this.hTableDescriptor.getTableName()
      , EventType.C_M_CREATE_TABLE.toString());
}

public DisableTableHandler(Server server, TableName tableName,
    AssignmentManager assignmentManager, TableLockManager tableLockManager,
    boolean skipTableStateCheck) {
  super(server, EventType.C_M_DISABLE_TABLE);
  this.tableName = tableName;
  this.assignmentManager = assignmentManager;
  this.tableLockManager = tableLockManager;
  this.skipTableStateCheck = skipTableStateCheck;
}

public EnableTableHandler(Server server, TableName tableName,
    AssignmentManager assignmentManager, TableLockManager tableLockManager,
    boolean skipTableStateCheck) {
  super(server, EventType.C_M_ENABLE_TABLE);
  this.tableName = tableName;
  this.assignmentManager = assignmentManager;
  this.tableLockManager = tableLockManager;
  this.skipTableStateCheck = skipTableStateCheck;
}

public ClosedRegionHandler(Server server, AssignmentManager assignmentManager,
    HRegionInfo regionInfo) {
  super(server, EventType.RS_ZK_REGION_CLOSED);
  this.assignmentManager = assignmentManager;
  this.regionInfo = regionInfo;
  if(regionInfo.isMetaRegion()) {
    priority = ClosedPriority.META;
  } else {
    priority = ClosedPriority.USER;
  }
}

public void offlineDisabledRegion(HRegionInfo regionInfo) {
  if (useZKForAssignment) {
    // Disabling so should not be reassigned, just delete the CLOSED node
    LOG.debug("Table being disabled so deleting ZK node and removing from " +
      "regions in transition, skipping assignment of region " +
        regionInfo.getRegionNameAsString());
    String encodedName = regionInfo.getEncodedName();
    deleteNodeInStates(encodedName, "closed", null,
      EventType.RS_ZK_REGION_CLOSED, EventType.M_ZK_REGION_OFFLINE);
  }
  replicasToClose.remove(regionInfo);
  regionOffline(regionInfo);
}

private void processAlreadyOpenedRegion(HRegionInfo region, ServerName sn) {
  // Remove region from in-memory transition and unassigned node from ZK
  // While trying to enable the table the regions of the table were
  // already enabled.
  LOG.debug("ALREADY_OPENED " + region.getRegionNameAsString()
    + " to " + sn);
  String encodedName = region.getEncodedName();

  //If use ZkForAssignment, region already Opened event should not be handled, 
  //leave it to zk event. See HBase-14407.
  if(useZKForAssignment){
    String node = ZKAssign.getNodeName(watcher, encodedName);
    Stat stat = new Stat();
    try {
      byte[] existingBytes = ZKUtil.getDataNoWatch(watcher, node, stat);
      if(existingBytes!=null){
        RegionTransition rt= RegionTransition.parseFrom(existingBytes);
        EventType et = rt.getEventType();
        if (et.equals(EventType.RS_ZK_REGION_OPENED)) {
          LOG.debug("ALREADY_OPENED " + region.getRegionNameAsString()
            + " and node in "+et+" state");
          return;
        }
      }
    } catch (KeeperException ke) {
      LOG.warn("Unexpected ZK exception getData " + node
        + " node for the region " + encodedName, ke);
    } catch (DeserializationException e) {
      LOG.warn("Get RegionTransition from zk deserialization failed! ", e);
    }

    deleteNodeInStates(encodedName, "offline", sn, EventType.M_ZK_REGION_OFFLINE);
  }

  regionStates.regionOnline(region, sn);
}

public OpenMetaHandler(final Server server,
    final RegionServerServices rsServices, HRegionInfo regionInfo,
    final HTableDescriptor htd, long masterSystemTime, OpenRegionCoordination coordination,
    OpenRegionCoordination.OpenRegionDetails ord) {
  super(server, rsServices, regionInfo, htd, EventType.M_RS_OPEN_META,
      masterSystemTime, coordination, ord);
}

public CloseRegionHandler(final Server server,
    final RegionServerServices rsServices,
    final HRegionInfo regionInfo, final boolean abort,
    CloseRegionCoordination closeRegionCoordination,
    CloseRegionCoordination.CloseRegionDetails crd,
    ServerName destination) {
  this(server, rsServices, regionInfo, abort, closeRegionCoordination, crd,
    EventType.M_RS_CLOSE_REGION, destination);
}

protected CloseRegionHandler(final Server server,
    final RegionServerServices rsServices, HRegionInfo regionInfo,
    boolean abort, CloseRegionCoordination closeRegionCoordination,
    CloseRegionCoordination.CloseRegionDetails crd,
    EventType eventType, ServerName destination) {
  super(server, eventType);
  this.server = server;
  this.rsServices = rsServices;
  this.regionInfo = regionInfo;
  this.abort = abort;
  this.destination = destination;
  this.closeRegionCoordination = closeRegionCoordination;
  this.closeRegionDetails = crd;
  useZKForAssignment = ConfigUtil.useZKForAssignment(server.getConfiguration());
}

public ParallelSeekHandler(KeyValueScanner scanner,Cell keyValue, 
    long readPoint, CountDownLatch latch) {
  super(null, EventType.RS_PARALLEL_SEEK);
  this.scanner = scanner;
  this.keyValue = keyValue;
  this.readPoint = readPoint;
  this.latch = latch;
}

public OpenRegionHandler(final Server server,
    final RegionServerServices rsServices, HRegionInfo regionInfo,
    HTableDescriptor htd, long masterSystemTime, OpenRegionCoordination coordination,
    OpenRegionCoordination.OpenRegionDetails ord) {
  this(server, rsServices, regionInfo, htd, EventType.M_RS_OPEN_REGION,
      masterSystemTime, coordination, ord);
}

protected OpenRegionHandler(final Server server,
    final RegionServerServices rsServices, final HRegionInfo regionInfo,
    final HTableDescriptor htd, EventType eventType, long masterSystemTime,
    OpenRegionCoordination coordination, OpenRegionCoordination.OpenRegionDetails ord) {
  super(server, eventType);
  this.rsServices = rsServices;
  this.regionInfo = regionInfo;
  this.htd = htd;
  this.coordination = coordination;
  this.ord = ord;
  useZKForAssignment = ConfigUtil.useZKForAssignment(server.getConfiguration());
  this.masterSystemTime = masterSystemTime;
}

public WALSplitterHandler(final Server server, SplitLogWorkerCoordination coordination,
    SplitLogWorkerCoordination.SplitTaskDetails splitDetails, CancelableProgressable reporter,
    AtomicInteger inProgressTasks, TaskExecutor splitTaskExecutor, RecoveryMode mode) {
  super(server, EventType.RS_LOG_REPLAY);
  this.splitTaskDetails = splitDetails;
  this.coordination = coordination;
  this.reporter = reporter;
  this.inProgressTasks = inProgressTasks;
  this.inProgressTasks.incrementAndGet();
  this.serverName = server.getServerName();
  this.splitTaskExecutor = splitTaskExecutor;
  this.mode = mode;
}

public CloseMetaHandler(final Server server,
    final RegionServerServices rsServices,
    final HRegionInfo regionInfo,
    final boolean abort, CloseRegionCoordination closeRegionCoordination,
    CloseRegionCoordination.CloseRegionDetails crd) {
  super(server, rsServices, regionInfo, abort, closeRegionCoordination,
    crd, EventType.M_RS_CLOSE_META);
}

public RegionReplicaFlushHandler(Server server, ClusterConnection connection,
    RpcRetryingCallerFactory rpcRetryingCallerFactory, RpcControllerFactory rpcControllerFactory,
    int operationTimeout, HRegion region) {
  super(server, EventType.RS_REGION_REPLICA_FLUSH);
  this.connection = connection;
  this.rpcRetryingCallerFactory = rpcRetryingCallerFactory;
  this.rpcControllerFactory = rpcControllerFactory;
  this.operationTimeout = operationTimeout;
  this.region = region;
}

public FinishRegionRecoveringHandler(RegionServerServices rss,
    String regionName, String path) {
  // we are using the open region handlers, since this operation is in the region open lifecycle
  super(rss, EventType.M_RS_OPEN_REGION);
  this.rss = rss;
  this.regionName = regionName;
  this.path = path;
}

private void setRegionOpenedOnZK(final ZooKeeperWatcher zkWatcher, final ServerName serverName,
                                 HRegionInfo hregionInfo) throws Exception {
  int version = ZKAssign.getVersion(zkWatcher, hregionInfo);
  int versionTransition = ZKAssign.transitionNode(zkWatcher,
      hregionInfo, serverName, EventType.M_ZK_REGION_OFFLINE,
      EventType.RS_ZK_REGION_OPENING, version);
  ZKAssign.transitionNodeOpened(zkWatcher, hregionInfo, serverName, versionTransition);
}

@Test (timeout=300000) public void testRestartClusterAfterKill()
throws Exception {
  UTIL.getConfiguration().setBoolean("hbase.assignment.usezk", true);
  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 = ServerName.valueOf(HMaster.MASTER, 1, System.currentTimeMillis());

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

  LOG.debug("Created UNASSIGNED zNode for ROOT and hbase: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.");
}

/**
 * Test if close region can handle ZK closing node version mismatch
 * @throws IOException
 * @throws NodeExistsException
 * @throws KeeperException
* @throws DeserializationException
 */
@Test public void testZKClosingNodeVersionMismatch()
throws IOException, NodeExistsException, KeeperException, DeserializationException {
  final Server server = new MockServer(HTU);
  final RegionServerServices rss = HTU.createMockRegionServerService();

  HTableDescriptor htd = TEST_HTD;
  final HRegionInfo hri = TEST_HRI;

  ZkCoordinatedStateManager coordinationProvider = new ZkCoordinatedStateManager();
  coordinationProvider.initialize(server);
  coordinationProvider.start();

  // open a region first so that it can be closed later
  OpenRegion(server, rss, htd, hri, coordinationProvider.getOpenRegionCoordination());

  // close the region
  // Create it CLOSING, which is what Master set before sending CLOSE RPC
  int versionOfClosingNode = ZKAssign.createNodeClosing(server.getZooKeeper(),
    hri, server.getServerName());

  // The CloseRegionHandler will validate the expected version
  // Given it is set to invalid versionOfClosingNode+1,
  // CloseRegionHandler should be M_ZK_REGION_CLOSING

  ZkCloseRegionCoordination.ZkCloseRegionDetails zkCrd =
    new ZkCloseRegionCoordination.ZkCloseRegionDetails();
  zkCrd.setPublishStatusInZk(true);
  zkCrd.setExpectedVersion(versionOfClosingNode+1);

  CloseRegionHandler handler = new CloseRegionHandler(server, rss, hri, false,
    coordinationProvider.getCloseRegionCoordination(), zkCrd);
  handler.process();

  // Handler should remain in M_ZK_REGION_CLOSING
  RegionTransition rt =
    RegionTransition.parseFrom(ZKAssign.getData(server.getZooKeeper(), hri.getEncodedName()));
  assertTrue(rt.getEventType().equals(EventType.M_ZK_REGION_CLOSING ));
}

/**
 * Test if the region can be closed properly
 * @throws IOException
 * @throws NodeExistsException
 * @throws KeeperException
* @throws org.apache.hadoop.hbase.exceptions.DeserializationException
 */
@Test public void testCloseRegion()
throws IOException, NodeExistsException, KeeperException, DeserializationException {
  final Server server = new MockServer(HTU);
  final RegionServerServices rss = HTU.createMockRegionServerService();

  HTableDescriptor htd = TEST_HTD;
  HRegionInfo hri = TEST_HRI;

  ZkCoordinatedStateManager coordinationProvider = new ZkCoordinatedStateManager();
  coordinationProvider.initialize(server);
  coordinationProvider.start();

  // open a region first so that it can be closed later
  OpenRegion(server, rss, htd, hri, coordinationProvider.getOpenRegionCoordination());

  // close the region
  // Create it CLOSING, which is what Master set before sending CLOSE RPC
  int versionOfClosingNode = ZKAssign.createNodeClosing(server.getZooKeeper(),
    hri, server.getServerName());

  // The CloseRegionHandler will validate the expected version
  // Given it is set to correct versionOfClosingNode,
  // CloseRegionHandlerit should be RS_ZK_REGION_CLOSED

  ZkCloseRegionCoordination.ZkCloseRegionDetails zkCrd =
    new ZkCloseRegionCoordination.ZkCloseRegionDetails();
  zkCrd.setPublishStatusInZk(true);
  zkCrd.setExpectedVersion(versionOfClosingNode);

  CloseRegionHandler handler = new CloseRegionHandler(server, rss, hri, false,
    coordinationProvider.getCloseRegionCoordination(), zkCrd);
  handler.process();
  // Handler should have transitioned it to RS_ZK_REGION_CLOSED
  RegionTransition rt = RegionTransition.parseFrom(
    ZKAssign.getData(server.getZooKeeper(), hri.getEncodedName()));
  assertTrue(rt.getEventType().equals(EventType.RS_ZK_REGION_CLOSED));
}

@Test
public void testFailedOpenRegion() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = HTU.createMockRegionServerService();

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

  ZkCoordinatedStateManager csm = new ZkCoordinatedStateManager();
  csm.initialize(server);
  csm.start();

  ZkOpenRegionCoordination.ZkOpenRegionDetails zkCrd =
    new ZkOpenRegionCoordination.ZkOpenRegionDetails();
  zkCrd.setServerName(server.getServerName());

  // Create the handler
  OpenRegionHandler handler =
    new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD, -1,
      csm.getOpenRegionCoordination(), zkCrd) {
      @Override
      HRegion openRegion() {
        // Fake failure of opening a region due to an IOE, which is caught
        return null;
      }
  };
  rsServices.getRegionsInTransitionInRS().put(
    TEST_HRI.getEncodedNameAsBytes(), Boolean.TRUE);
  handler.process();

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

@Test
public void testFailedUpdateMeta() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = HTU.createMockRegionServerService();

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

  // Create the handler
  ZkCoordinatedStateManager csm = new ZkCoordinatedStateManager();
  csm.initialize(server);
  csm.start();

  ZkOpenRegionCoordination.ZkOpenRegionDetails zkCrd =
    new ZkOpenRegionCoordination.ZkOpenRegionDetails();
  zkCrd.setServerName(server.getServerName());

  OpenRegionHandler handler = new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD,
    -1, csm.getOpenRegionCoordination(), zkCrd) {
      @Override
      boolean updateMeta(final HRegion r, long masterSystemTime) {
        // Fake failure of updating META
        return false;
      }
  };
  rsServices.getRegionsInTransitionInRS().put(
    TEST_HRI.getEncodedNameAsBytes(), Boolean.TRUE);
  handler.process();

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

@Test
public void testTransitionToFailedOpenEvenIfCleanupFails() throws Exception {
  Server server = new MockServer(HTU);
  RegionServerServices rsServices = HTU.createMockRegionServerService();
  // Create it OFFLINE, which is what it expects
  ZKAssign.createNodeOffline(server.getZooKeeper(), TEST_HRI, server.getServerName());
  // Create the handler
  ZkCoordinatedStateManager csm = new ZkCoordinatedStateManager();
  csm.initialize(server);
  csm.start();

  ZkOpenRegionCoordination.ZkOpenRegionDetails zkCrd =
    new ZkOpenRegionCoordination.ZkOpenRegionDetails();
  zkCrd.setServerName(server.getServerName());

  OpenRegionHandler handler = new OpenRegionHandler(server, rsServices, TEST_HRI, TEST_HTD,
    -1, csm.getOpenRegionCoordination(), zkCrd) {
    @Override
    boolean updateMeta(HRegion r, long masterSystemTime) {
      return false;
    };

    @Override
    void cleanupFailedOpen(HRegion region) throws IOException {
      throw new IOException("FileSystem got closed.");
    }
  };
  rsServices.getRegionsInTransitionInRS().put(TEST_HRI.getEncodedNameAsBytes(), Boolean.TRUE);
  try {
    handler.process();
  } catch (Exception e) {
    // Ignore the IOException that we have thrown from cleanupFailedOpen
  }
  RegionTransition rt = RegionTransition.parseFrom(ZKAssign.getData(server.getZooKeeper(),
      TEST_HRI.getEncodedName()));
  assertEquals(EventType.RS_ZK_REGION_FAILED_OPEN, rt.getEventType());
}