Java 类org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher 实例源码

/**
 * A quick test that hbase:meta is assigned; blocks for short time only.
 * @return True if hbase:meta location is available and verified as good.
 * @throws InterruptedException
 * @throws IOException
 */
private boolean isMetaAssignedQuickTest(final MasterProcedureEnv env)
throws InterruptedException, IOException {
  ZooKeeperWatcher zkw = env.getMasterServices().getZooKeeper();
  MetaTableLocator mtl = env.getMasterServices().getMetaTableLocator();
  boolean metaAssigned = false;
  // Is hbase:meta location available yet?
  if (mtl.isLocationAvailable(zkw)) {
    ClusterConnection connection = env.getMasterServices().getConnection();
    // Is hbase:meta location good yet?
    long timeout =
      env.getMasterConfiguration().getLong(KEY_SHORT_WAIT_ON_META, DEFAULT_SHORT_WAIT_ON_META);
    if (mtl.verifyMetaRegionLocation(connection, zkw, timeout)) {
      metaAssigned = true;
    }
  }
  return metaAssigned;
}

/**
 * Disable all archiving of files for a given table
 * <p>
 * Inherently an <b>asynchronous operation</b>.
 * @param zooKeeper watcher for the ZK cluster
 * @param table name of the table to disable
 * @throws KeeperException if an unexpected ZK connection issues occurs
 */
private void disable(ZooKeeperWatcher zooKeeper, byte[] table) throws KeeperException {
  // ensure the latest state of the archive node is found
  zooKeeper.sync(archiveZnode);

  // if the top-level archive node is gone, then we are done
  if (ZKUtil.checkExists(zooKeeper, archiveZnode) < 0) {
    return;
  }
  // delete the table node, from the archive
  String tableNode = this.getTableNode(table);
  // make sure the table is the latest version so the delete takes
  zooKeeper.sync(tableNode);

  LOG.debug("Attempting to delete table node:" + tableNode);
  ZKUtil.deleteNodeRecursively(zooKeeper, tableNode);
}

/**
 * Top-level watcher/controller for procedures across the cluster.
 * <p>
 * On instantiation, this ensures the procedure znodes exist.  This however requires the passed in
 *  watcher has been started.
 * @param watcher watcher for the cluster ZK. Owned by <tt>this</tt> and closed via
 *          {@link #close()}
 * @param procDescription name of the znode describing the procedure to run
 * @throws KeeperException when the procedure znodes cannot be created
 */
public ZKProcedureUtil(ZooKeeperWatcher watcher, String procDescription)
    throws KeeperException {
  super(watcher);
  // make sure we are listening for events
  watcher.registerListener(this);
  // setup paths for the zknodes used in procedures
  this.baseZNode = ZKUtil.joinZNode(watcher.baseZNode, procDescription);
  acquiredZnode = ZKUtil.joinZNode(baseZNode, ACQUIRED_BARRIER_ZNODE_DEFAULT);
  reachedZnode = ZKUtil.joinZNode(baseZNode, REACHED_BARRIER_ZNODE_DEFAULT);
  abortZnode = ZKUtil.joinZNode(baseZNode, ABORT_ZNODE_DEFAULT);

  // first make sure all the ZK nodes exist
  // make sure all the parents exist (sometimes not the case in tests)
  ZKUtil.createWithParents(watcher, acquiredZnode);
  // regular create because all the parents exist
  ZKUtil.createAndFailSilent(watcher, reachedZnode);
  ZKUtil.createAndFailSilent(watcher, abortZnode);
}

@Override
public Pair<ZKProcedureCoordinatorRpcs, List<ZKProcedureMemberRpcs>> start(
    ZooKeeperWatcher watcher, String operationName,
    ProcedureCoordinator coordinator, String controllerName,
    ProcedureMember member, List<String> expected) throws Exception {
  // start the controller
  ZKProcedureCoordinatorRpcs controller = new ZKProcedureCoordinatorRpcs(
      watcher, operationName, CONTROLLER_NODE_NAME);
  controller.start(coordinator);

  // make a cohort controller for each expected node

  List<ZKProcedureMemberRpcs> cohortControllers = new ArrayList<ZKProcedureMemberRpcs>();
  for (String nodeName : expected) {
    ZKProcedureMemberRpcs cc = new ZKProcedureMemberRpcs(watcher, operationName);
    cc.start(nodeName, member);
    cohortControllers.add(cc);
  }
  return new Pair<ZKProcedureCoordinatorRpcs, List<ZKProcedureMemberRpcs>>(
      controller, cohortControllers);
}

private void checkAndMigrateQueuesToPB(ZooKeeperWatcher zkw, String znode, String rs)
    throws KeeperException, NoNodeException, InterruptedException {
  String rsPath = ZKUtil.joinZNode(znode, rs);
  List<String> peers = ZKUtil.listChildrenNoWatch(zkw, rsPath);
  if (peers == null || peers.isEmpty()) return;
  String peerPath = null;
  for (String peer : peers) {
    peerPath = ZKUtil.joinZNode(rsPath, peer);
    List<String> files = ZKUtil.listChildrenNoWatch(zkw, peerPath);
    if (files == null || files.isEmpty()) continue;
    String filePath = null;
    for (String file : files) {
      filePath = ZKUtil.joinZNode(peerPath, file);
      byte[] data = ZKUtil.getData(zkw, filePath);
      if (data == null || Bytes.equals(data, HConstants.EMPTY_BYTE_ARRAY)) continue;
      if (ProtobufUtil.isPBMagicPrefix(data)) continue;
      ZKUtil.setData(zkw, filePath,
        ZKUtil.positionToByteArray(Long.parseLong(Bytes.toString(data))));
    }
  }
}

private void checkAndMigratePeerZnodesToPB(ZooKeeperWatcher zkw, String znode,
    List<String> peers) throws KeeperException, NoNodeException, InterruptedException {
  for (String peer : peers) {
    String peerZnode = ZKUtil.joinZNode(znode, peer);
    byte[] data = ZKUtil.getData(zkw, peerZnode);
    if (!ProtobufUtil.isPBMagicPrefix(data)) {
      migrateClusterKeyToPB(zkw, peerZnode, data);
    }
    String peerStatePath = ZKUtil.joinZNode(peerZnode,
      getConf().get("zookeeper.znode.replication.peers.state", "peer-state"));
    if (ZKUtil.checkExists(zkw, peerStatePath) != -1) {
      data = ZKUtil.getData(zkw, peerStatePath);
      if (ProtobufUtil.isPBMagicPrefix(data)) continue;
      migratePeerStateToPB(zkw, data, peerStatePath);
    }
  }
}

/**
 * Return the last failed RS name under /hbase/recovering-regions/encodedRegionName
 *
 * @param encodedRegionName
 * @throws KeeperException
 */
private String getLastFailedRSFromZK(String encodedRegionName) throws KeeperException {
  String result = null;
  long maxZxid = 0;
  ZooKeeperWatcher zkw = this.getZooKeeper();
  String nodePath = ZKUtil.joinZNode(zkw.recoveringRegionsZNode, encodedRegionName);
  List<String> failedServers = ZKUtil.listChildrenNoWatch(zkw, nodePath);
  if (failedServers == null || failedServers.isEmpty()) {
    return result;
  }
  for (String failedServer : failedServers) {
    String rsPath = ZKUtil.joinZNode(nodePath, failedServer);
    Stat stat = new Stat();
    ZKUtil.getDataNoWatch(zkw, rsPath, stat);
    if (maxZxid < stat.getCzxid()) {
      maxZxid = stat.getCzxid();
      result = failedServer;
    }
  }
  return result;
}

/**
 * Create a default snapshot handler - uses a zookeeper based member controller.
 * @param rss region server running the handler
 * @throws KeeperException if the zookeeper cluster cannot be reached
 */
@Override
public void initialize(RegionServerServices rss) throws KeeperException {
  this.rss = rss;
  ZooKeeperWatcher zkw = rss.getZooKeeper();
  this.memberRpcs = new ZKProcedureMemberRpcs(zkw,
      SnapshotManager.ONLINE_SNAPSHOT_CONTROLLER_DESCRIPTION);

  // read in the snapshot request configuration properties
  Configuration conf = rss.getConfiguration();
  long keepAlive = conf.getLong(SNAPSHOT_TIMEOUT_MILLIS_KEY, SNAPSHOT_TIMEOUT_MILLIS_DEFAULT);
  int opThreads = conf.getInt(SNAPSHOT_REQUEST_THREADS_KEY, SNAPSHOT_REQUEST_THREADS_DEFAULT);

  // create the actual snapshot procedure member
  ThreadPoolExecutor pool = ProcedureMember.defaultPool(rss.getServerName().toString(),
    opThreads, keepAlive);
  this.member = new ProcedureMember(memberRpcs, pool, new SnapshotSubprocedureBuilder());
}

@BeforeClass
public static void beforeClass() throws Exception {
  // setup configuration
  Configuration conf = UTIL.getConfiguration();
  SecureTestUtil.enableSecurity(conf);

  UTIL.startMiniCluster();

  // Wait for the ACL table to become available
  UTIL.waitTableEnabled(AccessControlLists.ACL_TABLE_NAME);

  ZKW = new ZooKeeperWatcher(UTIL.getConfiguration(),
    "TestTablePermissions", ABORTABLE);

  UTIL.createTable(TEST_TABLE, TEST_FAMILY);
  UTIL.createTable(TEST_TABLE2, TEST_FAMILY);
}

@Ignore @Test
public void testNamespaceJanitor() throws Exception {
  FileSystem fs = TEST_UTIL.getTestFileSystem();

  int fsCount = fs.listStatus(new Path(FSUtils.getRootDir(TEST_UTIL.getConfiguration()),
      HConstants.BASE_NAMESPACE_DIR)).length;
  Path fakeNSPath =
      FSUtils.getNamespaceDir(FSUtils.getRootDir(TEST_UTIL.getConfiguration()), "foo");
  assertTrue(fs.mkdirs(fakeNSPath));

  String fakeZnode = ZKUtil.joinZNode(ZooKeeperWatcher.namespaceZNode, "foo");
  int zkCount = ZKUtil.listChildrenNoWatch(TEST_UTIL.getZooKeeperWatcher(),
      ZooKeeperWatcher.namespaceZNode).size();
  ZKUtil.createWithParents(TEST_UTIL.getZooKeeperWatcher(), fakeZnode);
  Thread.sleep(10000);

  //verify namespace count is the same and orphan is removed
  assertFalse(fs.exists(fakeNSPath));
  assertEquals(fsCount, fs.listStatus(new Path(FSUtils.getRootDir(TEST_UTIL.getConfiguration()),
          HConstants.BASE_NAMESPACE_DIR)).length);

  assertEquals(-1, ZKUtil.checkExists(TEST_UTIL.getZooKeeperWatcher(), fakeZnode));
  assertEquals(zkCount,
      ZKUtil.listChildrenNoWatch(TEST_UTIL.getZooKeeperWatcher(),
          ZooKeeperWatcher.namespaceZNode).size());
}

@Test
public void testMultipleZK()
throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
  Table localMeta =
    new HTable(new Configuration(TEST_UTIL.getConfiguration()), TableName.META_TABLE_NAME);
  Configuration otherConf = new Configuration(TEST_UTIL.getConfiguration());
  otherConf.set(HConstants.ZOOKEEPER_QUORUM, "127.0.0.1");
  Table ipMeta = new HTable(otherConf, TableName.META_TABLE_NAME);

  // dummy, just to open the connection
  final byte [] row = new byte [] {'r'};
  localMeta.exists(new Get(row));
  ipMeta.exists(new Get(row));

  // make sure they aren't the same
  ZooKeeperWatcher z1 =
    getZooKeeperWatcher(HConnectionManager.getConnection(localMeta.getConfiguration()));
  ZooKeeperWatcher z2 =
    getZooKeeperWatcher(HConnectionManager.getConnection(otherConf));
  assertFalse(z1 == z2);
  assertFalse(z1.getQuorum().equals(z2.getQuorum()));

  localMeta.close();
  ipMeta.close();
}

/**
 * Create a znode with data
 * @throws Exception
 */
@Test
public void testCreateWithParents() throws Exception {
  ZooKeeperWatcher zkw =
      new ZooKeeperWatcher(new Configuration(TEST_UTIL.getConfiguration()),
          TestZooKeeper.class.getName(), null);
  byte[] expectedData = new byte[] { 1, 2, 3 };
  ZKUtil.createWithParents(zkw, "/l1/l2/l3/l4/testCreateWithParents", expectedData);
  byte[] data = ZKUtil.getData(zkw, "/l1/l2/l3/l4/testCreateWithParents");
  assertTrue(Bytes.equals(expectedData, data));
  ZKUtil.deleteNodeRecursively(zkw, "/l1");

  ZKUtil.createWithParents(zkw, "/testCreateWithParents", expectedData);
  data = ZKUtil.getData(zkw, "/testCreateWithParents");
  assertTrue(Bytes.equals(expectedData, data));
  ZKUtil.deleteNodeRecursively(zkw, "/testCreateWithParents");
}

/**
 * Create a bunch of znodes in a hierarchy, try deleting one that has childs (it will fail), then
 * delete it recursively, then delete the last znode
 * @throws Exception
 */
@Test
public void testZNodeDeletes() throws Exception {
  ZooKeeperWatcher zkw = new ZooKeeperWatcher(
    new Configuration(TEST_UTIL.getConfiguration()),
    TestZooKeeper.class.getName(), null);
  ZKUtil.createWithParents(zkw, "/l1/l2/l3/l4");
  try {
    ZKUtil.deleteNode(zkw, "/l1/l2");
    fail("We should not be able to delete if znode has childs");
  } catch (KeeperException ex) {
    assertNotNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2/l3/l4", null));
  }
  ZKUtil.deleteNodeRecursively(zkw, "/l1/l2");
  // make sure it really is deleted
  assertNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2/l3/l4", null));

  // do the same delete again and make sure it doesn't crash
  ZKUtil.deleteNodeRecursively(zkw, "/l1/l2");

  ZKUtil.deleteNode(zkw, "/l1");
  assertNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2", null));
}

private void testZNodeACLs() throws IOException, KeeperException, InterruptedException {

    ZooKeeperWatcher watcher = new ZooKeeperWatcher(conf, "IntegrationTestZnodeACLs", null);
    RecoverableZooKeeper zk = ZKUtil.connect(this.conf, watcher);

    String baseZNode = watcher.baseZNode;

    LOG.info("");
    LOG.info("***********************************************************************************");
    LOG.info("Checking ZK permissions, root znode: " + baseZNode);
    LOG.info("***********************************************************************************");
    LOG.info("");

    checkZnodePermsRecursive(watcher, zk, baseZNode);

    LOG.info("Checking ZK permissions: SUCCESS");
  }

private static ZKInterProcessReadWriteLock getReadWriteLock(String testName)
    throws IOException {
  MetadataHandler handler = new MetadataHandler() {
    @Override
    public void handleMetadata(byte[] ownerMetadata) {
      LOG.info("Lock info: " + Bytes.toString(ownerMetadata));
    }
  };
  ZooKeeperWatcher zkWatcher = getZooKeeperWatcher(testName);
  String znode = ZKUtil.joinZNode(zkWatcher.tableLockZNode, testName);

  return new ZKInterProcessReadWriteLock(zkWatcher, znode, handler);
}

public ZKVisibilityLabelWatcher(ZooKeeperWatcher watcher, VisibilityLabelsCache labelsCache,
    Configuration conf) {
  super(watcher);
  this.labelsCache = labelsCache;
  String labelZnodeParent = conf.get(VISIBILITY_LABEL_ZK_PATH, DEFAULT_VISIBILITY_LABEL_NODE);
  String userAuthsZnodeParent = conf.get(VISIBILITY_USER_AUTHS_ZK_PATH,
      DEFAULT_VISIBILITY_USER_AUTHS_NODE);
  this.labelZnode = ZKUtil.joinZNode(watcher.baseZNode, labelZnodeParent);
  this.userAuthsZnode = ZKUtil.joinZNode(watcher.baseZNode, userAuthsZnodeParent);
}

private VisibilityLabelsCache(ZooKeeperWatcher watcher, Configuration conf) throws IOException {
  zkVisibilityWatcher = new ZKVisibilityLabelWatcher(watcher, this, conf);
  try {
    zkVisibilityWatcher.start();
  } catch (KeeperException ke) {
    LOG.error("ZooKeeper initialization failed", ke);
    throw new IOException(ke);
  }
}

/**
 * @param sn Name of this mock regionserver
 * @throws IOException
 * @throws org.apache.hadoop.hbase.ZooKeeperConnectionException
 */
MockRegionServer(final Configuration conf, final ServerName sn)
throws ZooKeeperConnectionException, IOException {
  this.sn = sn;
  this.conf = conf;
  this.zkw = new ZooKeeperWatcher(conf, sn.toString(), this, true);
}

public ZKSecretWatcher(Configuration conf,
    ZooKeeperWatcher watcher,
    AuthenticationTokenSecretManager secretManager) {
  super(watcher);
  this.secretManager = secretManager;
  String keyZNodeParent = conf.get("zookeeper.znode.tokenauth.parent", DEFAULT_ROOT_NODE);
  this.baseKeyZNode = ZKUtil.joinZNode(watcher.baseZNode, keyZNodeParent);
  this.keysParentZNode = ZKUtil.joinZNode(baseKeyZNode, DEFAULT_KEYS_PARENT);
}

/**
 * Create a new secret manager instance for generating keys.
 * @param conf Configuration to use
 * @param zk Connection to zookeeper for handling leader elections
 * @param keyUpdateInterval Time (in milliseconds) between rolling a new master key for token signing
 * @param tokenMaxLifetime Maximum age (in milliseconds) before a token expires and is no longer valid
 */
/* TODO: Restrict access to this constructor to make rogues instances more difficult.
 * For the moment this class is instantiated from
 * org.apache.hadoop.hbase.ipc.SecureServer so public access is needed.
 */
public AuthenticationTokenSecretManager(Configuration conf,
    ZooKeeperWatcher zk, String serverName,
    long keyUpdateInterval, long tokenMaxLifetime) {
  this.zkWatcher = new ZKSecretWatcher(conf, zk, this);
  this.keyUpdateInterval = keyUpdateInterval;
  this.tokenMaxLifetime = tokenMaxLifetime;
  this.leaderElector = new LeaderElector(zk, serverName);
  this.name = NAME_PREFIX+serverName;
  this.clusterId = new ZKClusterId(zk, zk);
}

public LeaderElector(ZooKeeperWatcher watcher, String serverName) {
  setDaemon(true);
  setName("ZKSecretWatcher-leaderElector");
  zkLeader = new ZKLeaderManager(watcher,
      ZKUtil.joinZNode(zkWatcher.getRootKeyZNode(), "keymaster"),
      Bytes.toBytes(serverName), this);
}

/**
 * start a state tracker to check whether this peer is enabled or not
 *
 * @param zookeeper zk watcher for the local cluster
 * @param peerStateNode path to zk node which stores peer state
 * @throws KeeperException
 */
public void startStateTracker(ZooKeeperWatcher zookeeper, String peerStateNode)
    throws KeeperException {
  ensurePeerEnabled(zookeeper, peerStateNode);
  this.peerStateTracker = new PeerStateTracker(peerStateNode, zookeeper, this);
  this.peerStateTracker.start();
  try {
    this.readPeerStateZnode();
  } catch (DeserializationException e) {
    throw ZKUtil.convert(e);
  }
}

private static ZooKeeperWatcher newZooKeeperWatcher() throws IOException {
  return new ZooKeeperWatcher(UTIL.getConfiguration(), "testing utility", new Abortable() {
    @Override
    public void abort(String why, Throwable e) {
      throw new RuntimeException(
          "Unexpected abort in distributed three phase commit test:" + why, e);
    }

    @Override
    public boolean isAborted() {
      return false;
    }
  });
}

/**
 * Try to own the task by transitioning the zk node data from UNASSIGNED to OWNED.
 * <p>
 * This method is also used to periodically heartbeat the task progress by transitioning the node
 * from OWNED to OWNED.
 * <p>
 * @param isFirstTime shows whther it's the first attempt.
 * @param zkw zk wathcer
 * @param server name
 * @param task to own
 * @param taskZKVersion version of the task in zk
 * @return non-negative integer value when task can be owned by current region server otherwise -1
 */
protected static int attemptToOwnTask(boolean isFirstTime, ZooKeeperWatcher zkw,
    ServerName server, String task, RecoveryMode mode, int taskZKVersion) {
  int latestZKVersion = FAILED_TO_OWN_TASK;
  try {
    SplitLogTask slt = new SplitLogTask.Owned(server, mode);
    Stat stat = zkw.getRecoverableZooKeeper().setData(task, slt.toByteArray(), taskZKVersion);
    if (stat == null) {
      LOG.warn("zk.setData() returned null for path " + task);
      SplitLogCounters.tot_wkr_task_heartbeat_failed.incrementAndGet();
      return FAILED_TO_OWN_TASK;
    }
    latestZKVersion = stat.getVersion();
    SplitLogCounters.tot_wkr_task_heartbeat.incrementAndGet();
    return latestZKVersion;
  } catch (KeeperException e) {
    if (!isFirstTime) {
      if (e.code().equals(KeeperException.Code.NONODE)) {
        LOG.warn("NONODE failed to assert ownership for " + task, e);
      } else if (e.code().equals(KeeperException.Code.BADVERSION)) {
        LOG.warn("BADVERSION failed to assert ownership for " + task, e);
      } else {
        LOG.warn("failed to assert ownership for " + task, e);
      }
    }
  } catch (InterruptedException e1) {
    LOG.warn("Interrupted while trying to assert ownership of " + task + " "
        + StringUtils.stringifyException(e1));
    Thread.currentThread().interrupt();
  }
  SplitLogCounters.tot_wkr_task_heartbeat_failed.incrementAndGet();
  return FAILED_TO_OWN_TASK;
}

@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) {}

  // Create the master node with a dummy address
  ServerName master = ServerName.valueOf("localhost", 1, System.currentTimeMillis());
  // Should not have a master yet
  DummyMaster dummyMaster = new DummyMaster(zk,master);
  ClusterStatusTracker clusterStatusTracker =
    dummyMaster.getClusterStatusTracker();
  ActiveMasterManager activeMasterManager =
    dummyMaster.getActiveMasterManager();
  assertFalse(activeMasterManager.clusterHasActiveMaster.get());

  // First test becoming the active master uninterrupted
  MonitoredTask status = Mockito.mock(MonitoredTask.class);
  clusterStatusTracker.setClusterUp();

  activeMasterManager.blockUntilBecomingActiveMaster(100, status);
  assertTrue(activeMasterManager.clusterHasActiveMaster.get());
  assertMaster(zk, master);

  // Now pretend master restart
  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);
}

/**
 * Called by implementing classes.
 * @param zkWatcher
 * @param parentLockNode The lock ZNode path
 * @param metadata
 * @param handler
 * @param childNode The prefix for child nodes created under the parent
 */
protected ZKInterProcessLockBase(ZooKeeperWatcher zkWatcher,
    String parentLockNode, byte[] metadata, MetadataHandler handler, String childNode) {
  this.zkWatcher = zkWatcher;
  this.parentLockNode = parentLockNode;
  this.fullyQualifiedZNode = ZKUtil.joinZNode(parentLockNode, childNode);
  this.metadata = metadata;
  this.handler = handler;
  this.childZNode = childNode;
}

@Override
public void initialize(RegionServerServices rss) throws KeeperException {
  this.rss = rss;
  ZooKeeperWatcher zkw = rss.getZooKeeper();
  this.memberRpcs = new ZKProcedureMemberRpcs(zkw, getProcedureSignature());

  ThreadPoolExecutor pool =
      ProcedureMember.defaultPool(rss.getServerName().toString(), 1);
  this.member = new ProcedureMember(memberRpcs, pool, new SimleSubprocedureBuilder());
  LOG.info("Initialized: " + rss.getServerName().toString());
}

/**
 * Setup the config for the cluster
 */
@BeforeClass
public static void setupCluster() throws Exception {
  setupConf(UTIL.getConfiguration());
  UTIL.startMiniZKCluster();
  CONNECTION = (ClusterConnection)ConnectionFactory.createConnection(UTIL.getConfiguration());
  archivingClient = new ZKTableArchiveClient(UTIL.getConfiguration(), CONNECTION);
  // make hfile archiving node so we can archive files
  ZooKeeperWatcher watcher = UTIL.getZooKeeperWatcher();
  String archivingZNode = ZKTableArchiveClient.getArchiveZNode(UTIL.getConfiguration(), watcher);
  ZKUtil.createWithParents(watcher, archivingZNode);
}

OfflineCallback(final ZooKeeperWatcher zkw,
    final ServerName destination, final AtomicInteger counter,
    final Map<String, Integer> offlineNodesVersions) {
  this.callBack = new ExistCallback(
    destination, counter, offlineNodesVersions);
  this.destination = destination;
  this.counter = counter;
  this.zkw = zkw;
}

public HFileArchiveManager(HConnection connection, Configuration conf)
    throws ZooKeeperConnectionException, IOException {
  this.zooKeeper = new ZooKeeperWatcher(conf, "hfileArchiveManager-on-" + connection.toString(),
      connection);
  this.archiveZnode = ZKTableArchiveClient.getArchiveZNode(this.zooKeeper.getConfiguration(),
    this.zooKeeper);
}

/**
 * Perform a best effort enable of hfile retention, which relies on zookeeper communicating the //
 * * change back to the hfile cleaner.
 * <p>
 * No attempt is made to make sure that backups are successfully created - it is inherently an
 * <b>asynchronous operation</b>.
 * @param zooKeeper watcher connection to zk cluster
 * @param table table name on which to enable archiving
 * @throws KeeperException
 */
private void enable(ZooKeeperWatcher zooKeeper, byte[] table)
    throws KeeperException {
  LOG.debug("Ensuring archiving znode exists");
  ZKUtil.createAndFailSilent(zooKeeper, archiveZnode);

  // then add the table to the list of znodes to archive
  String tableNode = this.getTableNode(table);
  LOG.debug("Creating: " + tableNode + ", data: []");
  ZKUtil.createSetData(zooKeeper, tableNode, new byte[0]);
}

/**
 * create an address tracker instance
 * @param sn if not-null set the active master
 * @param infoPort if there is an active master, set its info port.
 */
private MasterAddressTracker setupMasterTracker(final ServerName sn, final int infoPort)
    throws Exception {
  ZooKeeperWatcher zk = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(),
      name.getMethodName(), null);
  ZKUtil.createAndFailSilent(zk, zk.baseZNode);

  // Should not have a master yet
  MasterAddressTracker addressTracker = new MasterAddressTracker(zk, null);
  addressTracker.start();
  assertFalse(addressTracker.hasMaster());
  zk.registerListener(addressTracker);

  // Use a listener to capture when the node is actually created
  NodeCreationListener listener = new NodeCreationListener(zk, zk.getMasterAddressZNode());
  zk.registerListener(listener);

  if (sn != null) {
    LOG.info("Creating master node");
    MasterAddressTracker.setMasterAddress(zk, zk.getMasterAddressZNode(), sn, infoPort);

    // Wait for the node to be created
    LOG.info("Waiting for master address manager to be notified");
    listener.waitForCreation();
    LOG.info("Master node created");
  }
  return addressTracker;
}

/**
 * @param htu Testing utility to use
 * @param zkw If true, create a zkw.
 * @throws ZooKeeperConnectionException
 * @throws IOException
 */
public MockServer(final HBaseTestingUtility htu, final boolean zkw)
throws ZooKeeperConnectionException, IOException {
  this.htu = htu;
  this.zk = zkw?
    new ZooKeeperWatcher(htu.getConfiguration(), NAME.toString(), this, true):
    null;
}

private ZooKeeperWatcher createZooKeeperWatcher() throws IOException {
  return new ZooKeeperWatcher(getConf(), "hbase Fsck", new Abortable() {
    @Override
    public void abort(String why, Throwable e) {
      LOG.error(why, e);
      System.exit(1);
    }

    @Override
    public boolean isAborted() {
      return false;
    }

  });
}

private ServerName getMetaRegionServerName(int replicaId)
throws IOException, KeeperException {
  ZooKeeperWatcher zkw = createZooKeeperWatcher();
  ServerName sn = null;
  try {
    sn = new MetaTableLocator().getMetaRegionLocation(zkw, replicaId);
  } finally {
    zkw.close();
  }
  return sn;
}

private void checkAndFixTableLocks() throws IOException {
  ZooKeeperWatcher zkw = createZooKeeperWatcher();

  try {
    TableLockChecker checker = new TableLockChecker(zkw, errors);
    checker.checkTableLocks();

    if (this.fixTableLocks) {
      checker.fixExpiredTableLocks();
    }
  } finally {
    zkw.close();
  }
}

private void unassignMetaReplica(HbckInfo hi) throws IOException, InterruptedException,
KeeperException {
  undeployRegions(hi);
  ZooKeeperWatcher zkw = createZooKeeperWatcher();
  try {
    ZKUtil.deleteNode(zkw, zkw.getZNodeForReplica(hi.metaEntry.getReplicaId()));
  } finally {
    zkw.close();
  }
}

@Test
public void testRemoveStaleRecoveringRegionsDuringMasterInitialization() throws Exception {
  // this test is for when distributed log replay is enabled
  if (!UTIL.getConfiguration().getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false)) return;

  LOG.info("Starting testRemoveStaleRecoveringRegionsDuringMasterInitialization");
  HMaster master = UTIL.getMiniHBaseCluster().getMaster();
  MasterFileSystem fs = master.getMasterFileSystem();

  String failedRegion = "failedRegoin1";
  String staleRegion = "staleRegion";
  ServerName inRecoveryServerName = ServerName.valueOf("mgr,1,1");
  ServerName previouselyFaildServerName = ServerName.valueOf("previous,1,1");
  String walPath = "/hbase/data/.logs/" + inRecoveryServerName.getServerName()
      + "-splitting/test";
  // Create a ZKW to use in the test
  ZooKeeperWatcher zkw = HBaseTestingUtility.getZooKeeperWatcher(UTIL);
  zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, walPath),
    new SplitLogTask.Owned(inRecoveryServerName, fs.getLogRecoveryMode()).toByteArray(), 
      Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  String staleRegionPath = ZKUtil.joinZNode(zkw.recoveringRegionsZNode, staleRegion);
  ZKUtil.createWithParents(zkw, staleRegionPath);
  String inRecoveringRegionPath = ZKUtil.joinZNode(zkw.recoveringRegionsZNode, failedRegion);
  inRecoveringRegionPath = ZKUtil.joinZNode(inRecoveringRegionPath, 
    inRecoveryServerName.getServerName());
  ZKUtil.createWithParents(zkw, inRecoveringRegionPath);
  Set<ServerName> servers = new HashSet<ServerName>();
  servers.add(previouselyFaildServerName);
  fs.removeStaleRecoveringRegionsFromZK(servers);

  // verification
  assertFalse(ZKUtil.checkExists(zkw, staleRegionPath) != -1);
  assertTrue(ZKUtil.checkExists(zkw, inRecoveringRegionPath) != -1);

  ZKUtil.deleteChildrenRecursively(zkw, zkw.recoveringRegionsZNode);
  ZKUtil.deleteChildrenRecursively(zkw, zkw.splitLogZNode);
  zkw.close();
}

public TableLockChecker(ZooKeeperWatcher zkWatcher, ErrorReporter errorReporter) {
  this.zkWatcher = zkWatcher;
  this.errorReporter = errorReporter;
  expireTimeout = zkWatcher.getConfiguration().getLong(
      TableLockManager.TABLE_LOCK_EXPIRE_TIMEOUT,
      TableLockManager.DEFAULT_TABLE_LOCK_EXPIRE_TIMEOUT_MS);
}

@Override
public String getZookeeperQuorum() {
  ZooKeeperWatcher zk = regionServer.getZooKeeper();
  if (zk == null) {
    return "";
  }
  return zk.getQuorum();
}