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

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");
  }

@Test (timeout=180000)
public void testTaskDone() throws Exception {
  LOG.info("TestTaskDone - cleanup task node once in DONE state");

  slm = new SplitLogManager(ds, conf, stopper, master, DUMMY_MASTER);
  TaskBatch batch = new TaskBatch();
  String tasknode = submitTaskAndWait(batch, "foo/1");
  final ServerName worker1 = ServerName.valueOf("worker1,1,1");
  SplitLogTask slt = new SplitLogTask.Done(worker1, this.mode);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  synchronized (batch) {
    while (batch.installed != batch.done) {
      batch.wait();
    }
  }
  waitForCounter(tot_mgr_task_deleted, 0, 1, to/2);
  assertTrue(ZKUtil.checkExists(zkw, tasknode) == -1);
}

private void refreshNodes(List<ZKUtil.NodeAndData> nodes) {
  for (ZKUtil.NodeAndData n : nodes) {
    String path = n.getNode();
    String keyId = ZKUtil.getNodeName(path);
    try {
      byte[] data = n.getData();
      if (data == null || data.length == 0) {
        LOG.debug("Ignoring empty node "+path);
        continue;
      }
      AuthenticationKey key = (AuthenticationKey)Writables.getWritable(
          data, new AuthenticationKey());
      secretManager.addKey(key);
    } catch (IOException ioe) {
      LOG.fatal("Failed reading new secret key for id '" + keyId +
          "' from zk", ioe);
      watcher.abort("Error deserializing key from znode "+path, ioe);
    }
  }
}

private void checkIfBaseNodeAvailable(ZooKeeperWatcher zkw)
  throws MasterNotRunningException {
  String errorMsg;
  try {
    if (ZKUtil.checkExists(zkw, zkw.baseZNode) == -1) {
      errorMsg = "The node " + zkw.baseZNode+" is not in ZooKeeper. "
        + "It should have been written by the master. "
        + "Check the value configured in 'zookeeper.znode.parent'. "
        + "There could be a mismatch with the one configured in the master.";
      LOG.error(errorMsg);
      throw new MasterNotRunningException(errorMsg);
    }
  } catch (KeeperException e) {
    errorMsg = "Can't get connection to ZooKeeper: " + e.getMessage();
    LOG.error(errorMsg);
    throw new MasterNotRunningException(errorMsg, e);
  }
}

@Test
public void testZookeeperNodesForReplicas() throws Exception {
  // Checks all the znodes exist when meta's replicas are enabled
  ZooKeeperWatcher zkw = TEST_UTIL.getZooKeeperWatcher();
  Configuration conf = TEST_UTIL.getConfiguration();
  String baseZNode = conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT,
      HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
  String primaryMetaZnode = ZKUtil.joinZNode(baseZNode,
      conf.get("zookeeper.znode.metaserver", "meta-region-server"));
  // check that the data in the znode is parseable (this would also mean the znode exists)
  byte[] data = ZKUtil.getData(zkw, primaryMetaZnode);
  ServerName.parseFrom(data);
  for (int i = 1; i < 3; i++) {
    String secZnode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.metaserver", "meta-region-server") + "-" + i);
    String str = zkw.getZNodeForReplica(i);
    assertTrue(str.equals(secZnode));
    // check that the data in the znode is parseable (this would also mean the znode exists)
    data = ZKUtil.getData(zkw, secZnode);
    ServerName.parseFrom(data);
  }
}

@Override
public void nodeDataChanged(String path) {
  waitUntilStarted();
  if (aclZNode.equals(ZKUtil.getParent(path))) {
    // update cache on an existing table node
    String entry = ZKUtil.getNodeName(path);
    try {
      byte[] data = ZKUtil.getDataAndWatch(watcher, path);
      refreshAuthManager(entry, data);
    } catch (KeeperException ke) {
      LOG.error("Error reading data from zookeeper for node " + entry, ke);
      // only option is to abort
      watcher.abort("Zookeeper error getting data for node " + entry, ke);
    } catch (IOException ioe) {
      LOG.error("Error reading permissions writables", ioe);
    }
  }
}

private void checkZnodePermsRecursive(ZooKeeperWatcher watcher,
    RecoverableZooKeeper zk, String znode) throws KeeperException, InterruptedException {

  boolean expectedWorldReadable = watcher.isClientReadable(znode);

  assertZnodePerms(zk, znode, expectedWorldReadable);

  try {
    List<String> children = zk.getChildren(znode, false);

    for (String child : children) {
      checkZnodePermsRecursive(watcher, zk, ZKUtil.joinZNode(znode, child));
    }
  } catch (KeeperException ke) {
    // if we are not authenticated for listChildren, it is fine.
    if (ke.code() != Code.NOAUTH) {
      throw ke;
    }
  }
}

/**
 * Creates a new ephemeral node in the PENDING_MERGE state for the merged region.
 * Create it ephemeral in case regionserver dies mid-merge.
 *
 * <p>
 * Does not transition nodes from other states. If a node already exists for
 * this region, a {@link org.apache.zookeeper.KeeperException.NodeExistsException} will be thrown.
 *
 * @param region region to be created as offline
 * @param serverName server event originates from
 * @throws IOException
 */
@Override
public void startRegionMergeTransaction(final HRegionInfo region, final ServerName serverName,
    final HRegionInfo a, final HRegionInfo b) throws IOException {
  LOG.debug(watcher.prefix("Creating ephemeral node for " + region.getEncodedName()
      + " in PENDING_MERGE state"));
  byte[] payload = HRegionInfo.toDelimitedByteArray(region, a, b);
  RegionTransition rt =
      RegionTransition.createRegionTransition(RS_ZK_REQUEST_REGION_MERGE, region.getRegionName(),
        serverName, payload);
  String node = ZKAssign.getNodeName(watcher, region.getEncodedName());
  try {
    if (!ZKUtil.createEphemeralNodeAndWatch(watcher, node, rt.toByteArray())) {
      throw new IOException("Failed create of ephemeral " + node);
    }
  } catch (KeeperException e) {
    throw new IOException(e);
  }
}

private List<String> getTaskList() throws InterruptedException {
  List<String> childrenPaths = null;
  long sleepTime = 1000;
  // It will be in loop till it gets the list of children or
  // it will come out if worker thread exited.
  while (!shouldStop) {
    try {
      childrenPaths =
          ZKUtil.listChildrenAndWatchForNewChildren(watcher,
            watcher.splitLogZNode);
      if (childrenPaths != null) {
        return childrenPaths;
      }
    } catch (KeeperException e) {
      LOG.warn("Could not get children of znode " + watcher.splitLogZNode, e);
    }
    LOG.debug("Retry listChildren of znode " + watcher.splitLogZNode
        + " after sleep for " + sleepTime + "ms!");
    Thread.sleep(sleepTime);
  }
  return childrenPaths;
}

@Override
public boolean isReady() throws InterruptedException {
  int result = -1;
  try {
    result = ZKUtil.checkExists(watcher, watcher.splitLogZNode);
  } catch (KeeperException e) {
    // ignore
    LOG.warn("Exception when checking for " + watcher.splitLogZNode
        + " ... retrying", e);
  }
  if (result == -1) {
    LOG.info(watcher.splitLogZNode
        + " znode does not exist, waiting for master to create");
    Thread.sleep(1000);
  }
  return (result != -1);
}

@Override
public int remainingTasksInCoordination() {
  int count = 0;
  try {
    List<String> tasks = ZKUtil.listChildrenNoWatch(watcher, watcher.splitLogZNode);
    if (tasks != null) {
      int listSize = tasks.size();
      for (int i = 0; i < listSize; i++) {
        if (!ZKSplitLog.isRescanNode(tasks.get(i))) {
          count++;
        }
      }
    }
  } catch (KeeperException ke) {
    LOG.warn("Failed to check remaining tasks", ke);
    count = -1;
  }
  return count;
}

public void start() throws IOException {
  watcher.registerListener(this);
  try {
    if (ZKUtil.watchAndCheckExists(watcher, nsZNode)) {
      List<ZKUtil.NodeAndData> existing =
          ZKUtil.getChildDataAndWatchForNewChildren(watcher, nsZNode);
      if (existing != null) {
        refreshNodes(existing);
      }
    } else {
      ZKUtil.createWithParents(watcher, nsZNode);
    }
  } catch (KeeperException e) {
    throw new IOException("Failed to initialize ZKNamespaceManager", e);
  }
}

@Test
public void testMasterRestartAfterNameSpaceEnablingNodeIsCreated() throws Exception {
  // Step 1: start mini zk cluster.
  MiniZooKeeperCluster zkCluster;
  zkCluster = TEST_UTIL.startMiniZKCluster();
  // Step 2: add an orphaned system table ZNODE
  TableName tableName = TableName.valueOf("hbase:namespace");
  ZooKeeperWatcher zkw = TEST_UTIL.getZooKeeperWatcher();
  String znode = ZKUtil.joinZNode(zkw.tableZNode, tableName.getNameAsString());
  ZooKeeperProtos.Table.Builder builder = ZooKeeperProtos.Table.newBuilder();
  builder.setState(ZooKeeperProtos.Table.State.ENABLED);
  byte [] data = ProtobufUtil.prependPBMagic(builder.build().toByteArray());
  ZKUtil.createSetData(zkw, znode, data);
  LOG.info("Create an orphaned Znode " + znode + " with data " + data);
  // Step 3: link the zk cluster to hbase cluster
  TEST_UTIL.setZkCluster(zkCluster);
  // Step 4: start hbase cluster and expect master to start successfully.
  TEST_UTIL.startMiniCluster();
  assertTrue(TEST_UTIL.getHBaseCluster().getLiveMasterThreads().size() == 1);
}

/**
 * {@inheritDoc}
 */
@Override
protected String getLockPath(String createdZNode, List<String> children) throws IOException {
  TreeSet<String> writeChildren =
      new TreeSet<String>(ZNodeComparator.COMPARATOR);
  for (String child : children) {
    if (isChildWriteLock(child)) {
      writeChildren.add(child);
    }
  }
  if (writeChildren.isEmpty()) {
    return null;
  }
  SortedSet<String> lowerChildren = writeChildren.headSet(createdZNode);
  if (lowerChildren.isEmpty()) {
    return null;
  }
  String pathToWatch = lowerChildren.last();
  String nodeHoldingLock = lowerChildren.first();
  String znode = ZKUtil.joinZNode(parentLockNode, nodeHoldingLock);
  handleLockMetadata(znode);

  return pathToWatch;
}

/**
 * Update state as to indicate that a lock is held
 * @param createdZNode The lock znode
 * @throws IOException If an unrecoverable ZooKeeper error occurs
 */
protected void updateAcquiredLock(String createdZNode) throws IOException {
  Stat stat = new Stat();
  byte[] data = null;
  Exception ex = null;
  try {
    data = ZKUtil.getDataNoWatch(zkWatcher, createdZNode, stat);
  } catch (KeeperException e) {
    LOG.warn("Cannot getData for znode:" + createdZNode, e);
    ex = e;
  }
  if (data == null) {
    LOG.error("Can't acquire a lock on a non-existent node " + createdZNode);
    throw new IllegalStateException("ZNode " + createdZNode +
        "no longer exists!", ex);
  }
  AcquiredLock newLock = new AcquiredLock(createdZNode, stat.getVersion());
  if (!acquiredLock.compareAndSet(null, newLock)) {
    LOG.error("The lock " + fullyQualifiedZNode +
        " has already been acquired by another process!");
    throw new IllegalStateException(fullyQualifiedZNode +
        " is held by another process");
  }
}

/**
 * Process metadata stored in a ZNode using a callback object passed to
 * this instance.
 * <p>
 * @param lockZNode The node holding the metadata
 * @param handler the metadata handler
 * @return True if metadata was ready and processed, false on exception.
 */
protected boolean handleLockMetadata(String lockZNode, MetadataHandler handler) {
  if (handler == null) {
    return false;
  }
  try {
    byte[] metadata = ZKUtil.getData(zkWatcher, lockZNode);
    handler.handleMetadata(metadata);
  } catch (KeeperException ex) {
    LOG.warn("Error processing lock metadata in " + lockZNode);
    return false;
  } catch (InterruptedException e) {
    LOG.warn("InterruptedException processing lock metadata in " + lockZNode);
    Thread.currentThread().interrupt();
    return false;
  }
  return true;
}

/**
 * Visits the locks (both held and attempted) with the given MetadataHandler.
 * @throws IOException If there is an unrecoverable error
 */
public void visitLocks(MetadataHandler handler) throws IOException {
  List<String> children;
  try {
    children = ZKUtil.listChildrenNoWatch(zkWatcher, parentLockNode);
  } catch (KeeperException e) {
    LOG.error("Unexpected ZooKeeper error when listing children", e);
    throw new IOException("Unexpected ZooKeeper exception", e);
  }
  if (children != null && children.size() > 0) {
    for (String child : children) {
      if (isChildOfSameType(child)) {
        String znode = ZKUtil.joinZNode(parentLockNode, child);
        String childWatchesZNode = getLockPath(child, children);
        if (childWatchesZNode == null) {
          LOG.info("Lock is held by: " + child);
        }
        handleLockMetadata(znode, handler);
      }
    }
  }
}

private InterProcessLock createTableLock() {
  String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode,
      tableName.getNameAsString());

  ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder()
    .setTableName(ProtobufUtil.toProtoTableName(tableName))
    .setLockOwner(ProtobufUtil.toServerName(serverName))
    .setThreadId(Thread.currentThread().getId())
    .setPurpose(purpose)
    .setIsShared(isShared)
    .setCreateTime(EnvironmentEdgeManager.currentTime()).build();
  byte[] lockMetadata = toBytes(data);

  InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode,
    METADATA_HANDLER);
  return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata);
}

/**
 * Sets znodes used in 0.94 version. Only table and replication znodes will be upgraded to PB,
 * others would be deleted.
 * @throws KeeperException
 */
private static void setUp94Znodes() throws IOException, KeeperException {
  // add some old znodes, which would be deleted after upgrade.
  String rootRegionServerZnode = ZKUtil.joinZNode(zkw.baseZNode, "root-region-server");
  ZKUtil.createWithParents(zkw, rootRegionServerZnode);
  ZKUtil.createWithParents(zkw, zkw.backupMasterAddressesZNode);
  // add table znode, data of its children would be protobuffized
  tableAZnode = ZKUtil.joinZNode(zkw.tableZNode, "a");
  ZKUtil.createWithParents(zkw, tableAZnode,
    Bytes.toBytes(ZooKeeperProtos.Table.State.ENABLED.toString()));
  // add replication znodes, data of its children would be protobuffized
  String replicationZnode = ZKUtil.joinZNode(zkw.baseZNode, "replication");
  replicationPeerZnode = ZKUtil.joinZNode(replicationZnode, "peers");
  peer1Znode = ZKUtil.joinZNode(replicationPeerZnode, "1");
  peer1 = ReplicationPeer.newBuilder().setClusterkey("abc:123:/hbase").build();
  ZKUtil.createWithParents(zkw, peer1Znode, Bytes.toBytes(peer1.getClusterkey()));
}

/**
 * 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");
}

/**
 * Handle a change in the master node.  Doesn't matter whether this was called
 * from a nodeCreated or nodeDeleted event because there are no guarantees
 * that the current state of the master node matches the event at the time of
 * our next ZK request.
 *
 * <p>Uses the watchAndCheckExists method which watches the master address node
 * regardless of whether it exists or not.  If it does exist (there is an
 * active master), it returns true.  Otherwise it returns false.
 *
 * <p>A watcher is set which guarantees that this method will get called again if
 * there is another change in the master node.
 */
private void handleMasterNodeChange() {
  // Watch the node and check if it exists.
  try {
    synchronized(clusterHasActiveMaster) {
      if (ZKUtil.watchAndCheckExists(watcher, watcher.getMasterAddressZNode())) {
        // A master node exists, there is an active master
        LOG.debug("A master is now available");
        clusterHasActiveMaster.set(true);
      } else {
        // Node is no longer there, cluster does not have an active master
        LOG.debug("No master available. Notifying waiting threads");
        clusterHasActiveMaster.set(false);
        // Notify any thread waiting to become the active master
        clusterHasActiveMaster.notifyAll();
      }
    }
  } catch (KeeperException ke) {
    master.abort("Received an unexpected KeeperException, aborting", ke);
  }
}

/**
 * Processes list of dead servers from result of hbase:meta scan and regions in RIT.
 * This is used for failover to recover the lost regions that belonged to
 * RegionServers which failed while there was no active master or are offline for whatever
 * reason and for regions that were in RIT.
 *
 * @param deadServers
 *          The list of dead servers which failed while there was no active master. Can be null.
 * @throws IOException
 * @throws KeeperException
 */
private void processDeadServersAndRecoverLostRegions(Set<ServerName> deadServers)
throws IOException, KeeperException {
  if (deadServers != null && !deadServers.isEmpty()) {
    for (ServerName serverName: deadServers) {
      if (!serverManager.isServerDead(serverName)) {
        serverManager.expireServer(serverName); // Let SSH do region re-assign
      }
    }
  }

  List<String> nodes = useZKForAssignment ?
    ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.assignmentZNode)
    : ZKUtil.listChildrenNoWatch(watcher, watcher.assignmentZNode);
  if (nodes != null && !nodes.isEmpty()) {
    for (String encodedRegionName : nodes) {
      processRegionInTransition(encodedRegionName, null);
    }
  } else if (!useZKForAssignment) {
    processRegionInTransitionZkLess();
  }
}

/**
 * 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);
}

/**
 * Sets the watch on the top-level archive znode, and then updates the monitor with the current
 * tables that should be archived (and ensures that those nodes are watched as well).
 */
private void checkEnabledAndUpdate() {
  try {
    if (ZKUtil.watchAndCheckExists(watcher, archiveHFileZNode)) {
      LOG.debug(archiveHFileZNode + " znode does exist, checking for tables to archive");

      // update the tables we should backup, to get the most recent state.
      // This is safer than also watching for children and then hoping we get
      // all the updates as it makes sure we get and watch all the children
      updateWatchedTables();
    } else {
      LOG.debug("Archiving not currently enabled, waiting");
    }
  } catch (KeeperException e) {
    LOG.warn("Failed to watch for archiving znode", e);
  }
}

@Before
public void setUp() throws Exception {
  zkw = HBaseTestingUtility.getZooKeeperWatcher(utility);
  String fakeRs1 = ZKUtil.joinZNode(zkw.rsZNode, "hostname1.example.org:1234");
  try {
    ZKClusterId.setClusterId(zkw, new ClusterId());
    rp = ReplicationFactory.getReplicationPeers(zkw, conf, zkw);
    rp.init();
    rt = ReplicationFactory.getReplicationTracker(zkw, rp, conf, zkw, new DummyServer(fakeRs1));
  } catch (Exception e) {
    fail("Exception during test setup: " + e);
  }
  rsRemovedCount = new AtomicInteger(0);
  rsRemovedData = "";
  plChangedCount = new AtomicInteger(0);
  plChangedData = new ArrayList<String>();
  peerRemovedCount = new AtomicInteger(0);
  peerRemovedData = "";
}

private void waitForNewProcedures() {
  // watch for new procedues that we need to start subprocedures for
  LOG.debug("Looking for new procedures under znode:'" + zkController.getAcquiredBarrier() + "'");
  List<String> runningProcedures = null;
  try {
    runningProcedures = ZKUtil.listChildrenAndWatchForNewChildren(zkController.getWatcher(),
      zkController.getAcquiredBarrier());
    if (runningProcedures == null) {
      LOG.debug("No running procedures.");
      return;
    }
  } catch (KeeperException e) {
    member.controllerConnectionFailure("General failure when watching for new procedures",
      e, null);
  }
  if (runningProcedures == null) {
    LOG.debug("No running procedures.");
    return;
  }
  for (String procName : runningProcedures) {
    // then read in the procedure information
    String path = ZKUtil.joinZNode(zkController.getAcquiredBarrier(), procName);
    startNewSubprocedure(path);
  }
}

/**
 * This should be called by the member and should write a serialized root cause exception as
 * to the abort znode.
 */
@Override
public void sendMemberAborted(Subprocedure sub, ForeignException ee) {
  if (sub == null) {
    LOG.error("Failed due to null subprocedure", ee);
    return;
  }
  String procName = sub.getName();
  LOG.debug("Aborting procedure (" + procName + ") in zk");
  String procAbortZNode = zkController.getAbortZNode(procName);
  try {
    String source = (ee.getSource() == null) ? memberName: ee.getSource();
    byte[] errorInfo = ProtobufUtil.prependPBMagic(ForeignException.serialize(source, ee));
    ZKUtil.createAndFailSilent(zkController.getWatcher(), procAbortZNode, errorInfo);
    LOG.debug("Finished creating abort znode:" + procAbortZNode);
  } catch (KeeperException e) {
    // possible that we get this error for the procedure if we already reset the zk state, but in
    // that case we should still get an error for that procedure anyways
    zkController.logZKTree(zkController.getBaseZnode());
    member.controllerConnectionFailure("Failed to post zk node:" + procAbortZNode
        + " to abort procedure", e, procName);
  }
}

/**
 * This is the abort message being sent by the coordinator to member
 *
 * TODO this code isn't actually used but can be used to issue a cancellation from the
 * coordinator.
 */
@Override
final public void sendAbortToMembers(Procedure proc, ForeignException ee) {
  String procName = proc.getName();
  LOG.debug("Aborting procedure '" + procName + "' in zk");
  String procAbortNode = zkProc.getAbortZNode(procName);
  try {
    LOG.debug("Creating abort znode:" + procAbortNode);
    String source = (ee.getSource() == null) ? coordName : ee.getSource();
    byte[] errorInfo = ProtobufUtil.prependPBMagic(ForeignException.serialize(source, ee));
    // first create the znode for the procedure
    ZKUtil.createAndFailSilent(zkProc.getWatcher(), procAbortNode, errorInfo);
    LOG.debug("Finished creating abort node:" + procAbortNode);
  } catch (KeeperException e) {
    // possible that we get this error for the procedure if we already reset the zk state, but in
    // that case we should still get an error for that procedure anyways
    zkProc.logZKTree(zkProc.baseZNode);
    coordinator.rpcConnectionFailure("Failed to post zk node:" + procAbortNode
        + " to abort procedure '" + procName + "'", new IOException(e));
  }
}

@Override
public void setPeerTableCFsConfig(String id, String tableCFsStr) throws ReplicationException {
  try {
    if (!peerExists(id)) {
      throw new IllegalArgumentException("Cannot set peer tableCFs because id=" + id
          + " does not exist.");
    }
    String tableCFsZKNode = getTableCFsNode(id);
    byte[] tableCFs = Bytes.toBytes(tableCFsStr);
    if (ZKUtil.checkExists(this.zookeeper, tableCFsZKNode) != -1) {
      ZKUtil.setData(this.zookeeper, tableCFsZKNode, tableCFs);
    } else {
      ZKUtil.createAndWatch(this.zookeeper, tableCFsZKNode, tableCFs);
    }
    LOG.info("Peer tableCFs with id= " + id + " is now " + tableCFsStr);
  } catch (KeeperException e) {
    throw new ReplicationException("Unable to change tableCFs of the peer with id=" + id, e);
  }
}

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))));
    }
  }
}

/**
 * Get the list of all the region servers from the specified peer
 * @param zkw zk connection to use
 * @return list of region server addresses or an empty list if the slave is unavailable
 */
protected static List<ServerName> fetchSlavesAddresses(ZooKeeperWatcher zkw)
    throws KeeperException {
  List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(zkw, zkw.rsZNode);
  if (children == null) {
    return Collections.emptyList();
  }
  List<ServerName> addresses = new ArrayList<ServerName>(children.size());
  for (String child : children) {
    addresses.add(ServerName.parseServerName(child));
  }
  return addresses;
}

public void start() throws KeeperException {
  watcher.registerListener(this);
  ZKUtil.createWithParents(watcher, labelZnode);
  ZKUtil.createWithParents(watcher, userAuthsZnode);
  byte[] data = ZKUtil.getDataAndWatch(watcher, labelZnode);
  if (data != null && data.length > 0) {
    refreshVisibilityLabelsCache(data);
  }
  data = ZKUtil.getDataAndWatch(watcher, userAuthsZnode);
  if (data != null && data.length > 0) {
    refreshUserAuthsCache(data);
  }
}

/**
 * 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;
}

/**
 * Write a labels mirror or user auths mirror into zookeeper
 * 
 * @param data
 * @param labelsOrUserAuths true for writing labels and false for user auths.
 */
public void writeToZookeeper(byte[] data, boolean labelsOrUserAuths) {
  String znode = this.labelZnode;
  if (!labelsOrUserAuths) {
    znode = this.userAuthsZnode;
  }
  try {
    ZKUtil.updateExistingNodeData(watcher, znode, data, -1);
  } catch (KeeperException e) {
    LOG.error("Failed writing to " + znode, e);
    watcher.abort("Failed writing node " + znode + " to zookeeper", e);
  }
}

public void start() throws KeeperException {
  watcher.registerListener(this);
  // make sure the base node exists
  ZKUtil.createWithParents(watcher, keysParentZNode);

  if (ZKUtil.watchAndCheckExists(watcher, keysParentZNode)) {
    List<ZKUtil.NodeAndData> nodes =
        ZKUtil.getChildDataAndWatchForNewChildren(watcher, keysParentZNode);
    refreshNodes(nodes);
  }
}

@Override
public void nodeDeleted(String path) {
  if (keysParentZNode.equals(ZKUtil.getParent(path))) {
    String keyId = ZKUtil.getNodeName(path);
    try {
      Integer id = Integer.valueOf(keyId);
      secretManager.removeKey(id);
    } catch (NumberFormatException nfe) {
      LOG.error("Invalid znode name for key ID '"+keyId+"'", nfe);
    }
  }
}

private void migratePeerStateToPB(ZooKeeperWatcher zkw, byte[] data,
String peerStatePath)
     throws KeeperException, NoNodeException {
   String state = Bytes.toString(data);
   if (ZooKeeperProtos.ReplicationState.State.ENABLED.name().equals(state)) {
     ZKUtil.setData(zkw, peerStatePath, ReplicationStateZKBase.ENABLED_ZNODE_BYTES);
   } else if (ZooKeeperProtos.ReplicationState.State.DISABLED.name().equals(state)) {
     ZKUtil.setData(zkw, peerStatePath, ReplicationStateZKBase.DISABLED_ZNODE_BYTES);
   }
 }

@Override
public void nodeChildrenChanged(String path) {
  if (path.equals(keysParentZNode)) {
    // keys changed
    try {
      List<ZKUtil.NodeAndData> nodes =
          ZKUtil.getChildDataAndWatchForNewChildren(watcher, keysParentZNode);
      refreshNodes(nodes);
    } catch (KeeperException ke) {
      LOG.fatal("Error reading data from zookeeper", ke);
      watcher.abort("Error reading changed keys from zookeeper", ke);
    }
  }
}

public void removeKeyFromZK(AuthenticationKey key) {
  String keyZNode = getKeyNode(key.getKeyId());
  try {
    ZKUtil.deleteNode(watcher, keyZNode);
  } catch (KeeperException.NoNodeException nne) {
    LOG.error("Non-existent znode "+keyZNode+" for key "+key.getKeyId(), nne);
  } catch (KeeperException ke) {
    LOG.fatal("Failed removing znode "+keyZNode+" for key "+key.getKeyId(),
        ke);
    watcher.abort("Unhandled zookeeper error removing znode "+keyZNode+
        " for key "+key.getKeyId(), ke);
  }
}

/**
 * refresh keys
 */
synchronized void refreshKeys() {
  try {
    List<ZKUtil.NodeAndData> nodes =
        ZKUtil.getChildDataAndWatchForNewChildren(watcher, keysParentZNode);
    refreshNodes(nodes);
  } catch (KeeperException ke) {
    LOG.fatal("Error reading data from zookeeper", ke);
    watcher.abort("Error reading changed keys from zookeeper", ke);
  }
}