Java 类org.apache.zookeeper.KeeperException 实例源码

private boolean createPath(String path, CreateMode createMode) {
    boolean success;
    try {
        zkClient.create()
                .creatingParentsIfNeeded()
                .withMode(createMode)
                .forPath(path, "".getBytes());
        success = true;
        LOG.info("create path success, path={}", path);
    } catch (KeeperException.NodeExistsException ex1) {
        success = true;
        LOG.debug("node exist, path={}", path);
    } catch (Exception ex2) {
        success = false;
        LOG.debug("createPath exception:", ex2);
    }
    return success;
}

/**
 * If region is up in zk in transition, then do fixup and block and wait until
 * the region is assigned and out of transition.  Used on startup for
 * catalog regions.
 * @param hri Region to look for.
 * @return True if we processed a region in transition else false if region
 * was not up in zk in transition.
 * @throws InterruptedException
 * @throws KeeperException
 * @throws IOException
 */
boolean processRegionInTransitionAndBlockUntilAssigned(final HRegionInfo hri)
    throws InterruptedException, KeeperException, IOException {
  String encodedRegionName = hri.getEncodedName();
  if (!processRegionInTransition(encodedRegionName, hri)) {
    return false; // The region is not in transition
  }
  LOG.debug("Waiting on " + HRegionInfo.prettyPrint(encodedRegionName));
  while (!this.server.isStopped() &&
      this.regionStates.isRegionInTransition(encodedRegionName)) {
    RegionState state = this.regionStates.getRegionTransitionState(encodedRegionName);
    if (state == null || !serverManager.isServerOnline(state.getServerName())) {
      // The region is not in transition, or not in transition on an online
      // server. Doesn't help to block here any more. Caller need to
      // verify the region is actually assigned.
      break;
    }
    this.regionStates.waitForUpdate(100);
  }
  return true;
}

@Override
public void processRequest(Request request)
        throws RequestProcessorException {
    // Check if this is a local session and we are trying to create
    // an ephemeral node, in which case we upgrade the session
    Request upgradeRequest = null;
    try {
        upgradeRequest = lzks.checkUpgradeSession(request);
    } catch (KeeperException ke) {
        if (request.getHdr() != null) {
            LOG.debug("Updating header");
            request.getHdr().setType(OpCode.error);
            request.setTxn(new ErrorTxn(ke.code().intValue()));
        }
        request.setException(ke);
        LOG.info("Error creating upgrade request " + ke.getMessage());
    } catch (IOException ie) {
        LOG.error("Unexpected error in upgrade", ie);
    }
    if (upgradeRequest != null) {
        nextProcessor.processRequest(upgradeRequest);
    }

    nextProcessor.processRequest(request);
}

/**
 * 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 the specified node with the specified data and watches it.
 *
 * <p>Throws an exception if the node already exists.
 *
 * <p>The node created is persistent and open access.
 *
 * <p>Returns the version number of the created node if successful.
 *
 * @param zkw zk reference
 * @param znode path of node to create
 * @param data data of node to create
 * @return version of node created
 * @throws KeeperException if unexpected zookeeper exception
 * @throws KeeperException.NodeExistsException if node already exists
 */
public static int createAndWatch(ZooKeeperWatcher zkw,
    String znode, byte [] data)
throws KeeperException, KeeperException.NodeExistsException {
  try {
    zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
        CreateMode.PERSISTENT);
    Stat stat = zkw.getRecoverableZooKeeper().exists(znode, zkw);
    if (stat == null){
      // Likely a race condition. Someone deleted the znode.
      throw KeeperException.create(KeeperException.Code.SYSTEMERROR,
          "ZK.exists returned null (i.e.: znode does not exist) for znode=" + znode);
    }
   return stat.getVersion();
  } catch (InterruptedException e) {
    zkw.interruptedException(e);
    return -1;
  }
}

private void createZNodeTree(String rootZNode) throws KeeperException,
    InterruptedException {
  List<Op> opList = new ArrayList<Op>();
  opList.add(Op.create(rootZNode, new byte[0], Ids.OPEN_ACL_UNSAFE,
      CreateMode.PERSISTENT));
  int level = 0;
  String parentZNode = rootZNode;
  while (level < 10) {
    // define parent node
    parentZNode = parentZNode + "/" + level;
    opList.add(Op.create(parentZNode, new byte[0], Ids.OPEN_ACL_UNSAFE,
        CreateMode.PERSISTENT));
    int elements = 0;
    // add elements to the parent node
    while (elements < level) {
      opList.add(Op.create(parentZNode + "/" + elements, new byte[0],
          Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
      elements++;
    }
    level++;
  }
  zkw.getRecoverableZooKeeper().multi(opList);
}

@Test(timeout = 10000)
public void testReconfigEnabledWithAuthAndWrongACL() throws InterruptedException {
    resetZKAdmin();

    try {
        zkAdmin.addAuthInfo("digest", "super:test".getBytes());
        // There is ACL however the permission is wrong - need WRITE permission at leaste.
        ArrayList<ACL> acls = new ArrayList<ACL>(
                Collections.singletonList(
                        new ACL(ZooDefs.Perms.READ,
                                new Id("digest", "user:tl+z3z0vO6PfPfEENfLF96E6pM0="/* password is test */))));
        zkAdmin.setACL(ZooDefs.CONFIG_NODE, acls, -1);
        resetZKAdmin();
        zkAdmin.addAuthInfo("digest", "user:test".getBytes());
        reconfigPort();
        Assert.fail("Reconfig should fail with an ACL that is read only!");
    } catch (KeeperException e) {
        Assert.assertTrue(e.code() == KeeperException.Code.NOAUTH);
    }
}

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

public static void deleteRecoveringRegionZNodes(ZooKeeperWatcher watcher, List<String> regions) {
  try {
    if (regions == null) {
      // remove all children under /home/recovering-regions
      LOG.debug("Garbage collecting all recovering region znodes");
      ZKUtil.deleteChildrenRecursively(watcher, watcher.recoveringRegionsZNode);
    } else {
      for (String curRegion : regions) {
        String nodePath = ZKUtil.joinZNode(watcher.recoveringRegionsZNode, curRegion);
        ZKUtil.deleteNodeRecursively(watcher, nodePath);
      }
    }
  } catch (KeeperException e) {
    LOG.warn("Cannot remove recovering regions from ZooKeeper", e);
  }
}

@DELETE
@Produces( { MediaType.APPLICATION_JSON, "application/javascript",
        MediaType.APPLICATION_XML, MediaType.APPLICATION_OCTET_STREAM })
public void deleteZNode(@PathParam("path") String path,
        @DefaultValue("-1") @QueryParam("version") String versionParam,
        @Context UriInfo ui) throws InterruptedException, KeeperException {
    ensurePathNotNull(path);

    int version;
    try {
        version = Integer.parseInt(versionParam);
    } catch (NumberFormatException e) {
        throw new WebApplicationException(Response.status(
                Response.Status.BAD_REQUEST).entity(
                new ZError(ui.getRequestUri().toString(), path
                        + " bad version " + versionParam)).build());
    }

    zk.delete(path, version);
}

@Test(timeout = 10000)
public void testReconfigEnabledWithAuthAndACL() throws InterruptedException {
    resetZKAdmin();

    try {
        zkAdmin.addAuthInfo("digest", "super:test".getBytes());
        ArrayList<ACL> acls = new ArrayList<ACL>(
                Collections.singletonList(
                        new ACL(ZooDefs.Perms.WRITE,
                        new Id("digest", "user:tl+z3z0vO6PfPfEENfLF96E6pM0="/* password is test */))));
        zkAdmin.setACL(ZooDefs.CONFIG_NODE, acls, -1);
        resetZKAdmin();
        zkAdmin.addAuthInfo("digest", "user:test".getBytes());
        Assert.assertTrue(reconfigPort());
    } catch (KeeperException e) {
        Assert.fail("Reconfig should not fail, but failed with exception : " + e.getMessage());
    }
}

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

@Override
public Stat setData(String path, byte[] data, int version)
        throws KeeperException, InterruptedException {
    int count = 0;
    do {
        try {
            return super.setData(path, data, version);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
            Stat s = exists(path, false);
            if (s != null) {
                if (getData(path, false, s) == data) {
                    return s;
                }
            } else {
                return null;
            }
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return null;
}

/**
 * Test verifies the multi calls with blank znode path
 */
@Test(timeout = 90000)
public void testBlankPath() throws Exception {
    List<Integer> expectedResultCodes = new ArrayList<Integer>();
    expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
                            .intValue());
    expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
    expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
                            .intValue());
    expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());

    // delete
    String expectedErr = "Path cannot be null";
    List<Op> opList = Arrays.asList(Op.delete("/multi0", -1),
                                    Op.delete(null, 100),
                                    Op.delete("/multi2", 5),
                                    Op.delete("", -1));
    multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
}

@Test
public void testNodes3() throws IOException, InterruptedException,
    KeeperException {

  int testIterations = 3;
  final CountDownLatch latch = new CountDownLatch(testIterations);
  final AtomicInteger failureCounter = new AtomicInteger();

  for (int i = 0; i < testIterations; i++) {
    runElectionSupportThread(latch, failureCounter);
  }

  Assert.assertEquals(0, failureCounter.get());

  if (!latch.await(10, TimeUnit.SECONDS)) {
    logger
        .info(
            "Waited for all threads to start, but timed out. We had {} failures.",
            failureCounter);
  }
}

/**
 * Creates the specified node and all parent nodes required for it to exist.  The creation of
 * parent znodes is not atomic with the leafe znode creation but the data is written atomically
 * when the leaf node is created.
 *
 * No watches are set and no errors are thrown if the node already exists.
 *
 * The nodes created are persistent and open access.
 *
 * @param zkw zk reference
 * @param znode path of node
 * @throws KeeperException if unexpected zookeeper exception
 */
public static void createWithParents(ZooKeeperWatcher zkw, String znode, byte[] data)
throws KeeperException {
  try {
    if(znode == null) {
      return;
    }
    zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
        CreateMode.PERSISTENT);
  } catch(KeeperException.NodeExistsException nee) {
    return;
  } catch(KeeperException.NoNodeException nne) {
    createWithParents(zkw, getParent(znode));
    createWithParents(zkw, znode, data);
  } catch(InterruptedException ie) {
    zkw.interruptedException(ie);
  }
}

@Test
public void testNullData() throws IOException, 
    InterruptedException, KeeperException {
    String path = "/SIZE";
    ZooKeeper zk = null;
    zk = createClient();
    try {
        zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        // try sync zk exists 
        zk.exists(path, false);
        zk.exists(path, false, this , null);
        cn.await(10, TimeUnit.SECONDS);
        Assert.assertSame(0L, cn.getCount());
    } finally {
        if(zk != null)
            zk.close();
    }

}

/**
 * This attempts to create an acquired state znode for the procedure (snapshot name).
 *
 * It then looks for the reached znode to trigger in-barrier execution.  If not present we
 * have a watcher, if present then trigger the in-barrier action.
 */
@Override
public void sendMemberAcquired(Subprocedure sub) throws IOException {
  String procName = sub.getName();
  try {
    LOG.debug("Member: '" + memberName + "' joining acquired barrier for procedure (" + procName
        + ") in zk");
    String acquiredZNode = ZKUtil.joinZNode(ZKProcedureUtil.getAcquireBarrierNode(
      zkController, procName), memberName);
    ZKUtil.createAndFailSilent(zkController.getWatcher(), acquiredZNode);

    // watch for the complete node for this snapshot
    String reachedBarrier = zkController.getReachedBarrierNode(procName);
    LOG.debug("Watch for global barrier reached:" + reachedBarrier);
    if (ZKUtil.watchAndCheckExists(zkController.getWatcher(), reachedBarrier)) {
      receivedReachedGlobalBarrier(reachedBarrier);
    }
  } catch (KeeperException e) {
    member.controllerConnectionFailure("Failed to acquire barrier for procedure: "
        + procName + " and member: " + memberName, e, procName);
  }
}

/**
 * Helper method that creates fencing node, executes the passed
 * delete related operations and deletes the fencing node.
 */
private synchronized void doDeleteMultiWithRetries(
    final List<Op> opList) throws Exception {
  final List<Op> execOpList = new ArrayList<Op>(opList.size() + 2);
  execOpList.add(createFencingNodePathOp);
  execOpList.addAll(opList);
  execOpList.add(deleteFencingNodePathOp);
  new ZKAction<Void>() {
    @Override
    public Void run() throws KeeperException, InterruptedException {
      setHasDeleteNodeOp(true);
      zkClient.multi(execOpList);
      return null;
    }
  }.runWithRetries();
}

public Stat setData(String path, byte data[], int version, long zxid,
        long time) throws KeeperException.NoNodeException {
    Stat s = new Stat();
    DataNode n = nodes.get(path);
    if (n == null) {
        throw new KeeperException.NoNodeException();
    }
    byte lastdata[] = null;
    synchronized (n) {
        lastdata = n.data;
        n.data = data;
        n.stat.setMtime(time);
        n.stat.setMzxid(zxid);
        n.stat.setVersion(version);
        n.copyStat(s);
    }
    // now update if the path is in a quota subtree.
    String lastPrefix = getMaxPrefixWithQuota(path);
    if(lastPrefix != null) {
      this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
          - (lastdata == null ? 0 : lastdata.length));
    }
    dataWatches.triggerWatch(path, EventType.NodeDataChanged);
    return s;
}

/**
 * exists is an idempotent operation. Retry before throwing exception
 * @return A Stat instance
 */
public Stat exists(String path, boolean watch)
throws KeeperException, InterruptedException {
  TraceScope traceScope = null;
  try {
    traceScope = Trace.startSpan("RecoverableZookeeper.exists");
    RetryCounter retryCounter = retryCounterFactory.create();
    while (true) {
      try {
        return checkZk().exists(path, watch);
      } catch (KeeperException e) {
        switch (e.code()) {
          case CONNECTIONLOSS:
          case SESSIONEXPIRED:
          case OPERATIONTIMEOUT:
            retryOrThrow(retryCounter, e, "exists");
            break;

          default:
            throw e;
        }
      }
      retryCounter.sleepUntilNextRetry();
    }
  } finally {
    if (traceScope != null) traceScope.close();
  }
}

/**
 * If there is a breadcrumb node indicating that another node may need
 * fencing, try to fence that node.
 * @return the Stat of the breadcrumb node that was read, or null
 * if no breadcrumb node existed
 */
private Stat fenceOldActive() throws InterruptedException, KeeperException {
  final Stat stat = new Stat();
  byte[] data;
  LOG.info("Checking for any old active which needs to be fenced...");
  try {
    data = zkDoWithRetries(new ZKAction<byte[]>() {
      @Override
      public byte[] run() throws KeeperException, InterruptedException {
        return zkClient.getData(zkBreadCrumbPath, false, stat);
      }
    });
  } catch (KeeperException ke) {
    if (isNodeDoesNotExist(ke.code())) {
      LOG.info("No old node to fence");
      return null;
    }

    // If we failed to read for any other reason, then likely we lost
    // our session, or we don't have permissions, etc. In any case,
    // we probably shouldn't become active, and failing the whole
    // thing is the best bet.
    throw ke;
  }

  LOG.info("Old node exists: " + StringUtils.byteToHexString(data));
  if (Arrays.equals(data, appData)) {
    LOG.info("But old node has our own data, so don't need to fence it.");
  } else {
    appClient.fenceOldActive(data);
  }
  return stat;
}

/**
 * Delete all the replication queues for a given region server.
 * @param regionserverZnode The znode of the region server to delete.
 */
private void deleteAnotherRSQueues(String regionserverZnode) {
  String fullpath = ZKUtil.joinZNode(this.queuesZNode, regionserverZnode);
  try {
    List<String> clusters = ZKUtil.listChildrenNoWatch(this.zookeeper, fullpath);
    for (String cluster : clusters) {
      // No need to delete, it will be deleted later.
      if (cluster.equals(RS_LOCK_ZNODE)) {
        continue;
      }
      String fullClusterPath = ZKUtil.joinZNode(fullpath, cluster);
      ZKUtil.deleteNodeRecursively(this.zookeeper, fullClusterPath);
    }
    // Finish cleaning up
    ZKUtil.deleteNodeRecursively(this.zookeeper, fullpath);
  } catch (KeeperException e) {
    if (e instanceof KeeperException.NoNodeException
        || e instanceof KeeperException.NotEmptyException) {
      // Testing a special case where another region server was able to
      // create a lock just after we deleted it, but then was also able to
      // delete the RS znode before us or its lock znode is still there.
      if (e.getPath().equals(fullpath)) {
        return;
      }
    }
    this.abortable.abort("Failed to delete replication queues for region server: "
        + regionserverZnode, e);
  }
}

@Test
public void testReplicationQueuesClient() throws ReplicationException, KeeperException {
  rqc.init();
  // Test methods with empty state
  assertEquals(0, rqc.getListOfReplicators().size());
  assertNull(rqc.getLogsInQueue(server1, "qId1"));
  assertNull(rqc.getAllQueues(server1));

  /*
   * Set up data Two replicators: -- server1: three queues with 0, 1 and 2 log files each --
   * server2: zero queues
   */
  rq1.init(server1);
  rq2.init(server2);
  rq1.addLog("qId1", "trash");
  rq1.removeLog("qId1", "trash");
  rq1.addLog("qId2", "filename1");
  rq1.addLog("qId3", "filename2");
  rq1.addLog("qId3", "filename3");
  rq2.addLog("trash", "trash");
  rq2.removeQueue("trash");

  List<String> reps = rqc.getListOfReplicators();
  assertEquals(2, reps.size());
  assertTrue(server1, reps.contains(server1));
  assertTrue(server2, reps.contains(server2));

  assertNull(rqc.getLogsInQueue("bogus", "bogus"));
  assertNull(rqc.getLogsInQueue(server1, "bogus"));
  assertEquals(0, rqc.getLogsInQueue(server1, "qId1").size());
  assertEquals(1, rqc.getLogsInQueue(server1, "qId2").size());
  assertEquals("filename1", rqc.getLogsInQueue(server1, "qId2").get(0));

  assertNull(rqc.getAllQueues("bogus"));
  assertEquals(0, rqc.getAllQueues(server2).size());
  List<String> list = rqc.getAllQueues(server1);
  assertEquals(3, list.size());
  assertTrue(list.contains("qId2"));
  assertTrue(list.contains("qId3"));
}

public static String reconfig(ZooKeeperAdmin zkAdmin, List<String> joiningServers,
                              List<String> leavingServers, List<String> newMembers, long fromConfig)
        throws KeeperException, InterruptedException {
    byte[] config = null;
    for (int j = 0; j < 30; j++) {
        try {
            config = zkAdmin.reconfigure(joiningServers, leavingServers,
                    newMembers, fromConfig, new Stat());
            break;
        } catch (KeeperException.ConnectionLossException e) {
            if (j < 29) {
                Thread.sleep(1000);
            } else {
                // test fails if we still can't connect to the quorum after
                // 30 seconds.
                Assert.fail("client could not connect to reestablished quorum: giving up after 30+ seconds.");
            }
        }
    }

    String configStr = new String(config);
    if (joiningServers != null) {
        for (String joiner : joiningServers)
            Assert.assertTrue(configStr.contains(joiner));
    }
    if (leavingServers != null) {
        for (String leaving : leavingServers)
            Assert.assertFalse(configStr.contains("server.".concat(leaving)));
    }

    return configStr;
}

/**
 * Start monitoring for archive updates
 * @throws KeeperException on failure to find/create nodes
 */
public void start() throws KeeperException {
  // if archiving is enabled, then read in the list of tables to archive
  LOG.debug("Starting hfile archive tracker...");
  this.checkEnabledAndUpdate();
  LOG.debug("Finished starting hfile archive tracker!");
}

@Override
public List<String> getChildren(String path, boolean watch)
        throws KeeperException, InterruptedException {
    int count = 0;
    do {
        try {
            return super.getChildren(path, watch ? watcher : null);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return new ArrayList<String>();
}

@Override
public synchronized void nodeCreated(String path) {
  if (!path.equals(node)) return;
  try {
    byte [] data = ZKUtil.getDataAndWatch(watcher, node);
    if (data != null) {
      this.data = data;
      notifyAll();
    } else {
      nodeDeleted(path);
    }
  } catch(KeeperException e) {
    abortable.abort("Unexpected exception handling nodeCreated event", e);
  }
}

private void createMyEphemeralNode() throws KeeperException, IOException {
  RegionServerInfo.Builder rsInfo = RegionServerInfo.newBuilder();
  rsInfo.setInfoPort(infoServer != null ? infoServer.getPort() : -1);
  rsInfo.setVersionInfo(ProtobufUtil.getVersionInfo());
  byte[] data = ProtobufUtil.prependPBMagic(rsInfo.build().toByteArray());
  ZKUtil.createEphemeralNodeAndWatch(this.zooKeeper, getMyEphemeralNodePath(), data);
}

public void my_test_1() throws IOException,
        InterruptedException, KeeperException {
  enode_test_1();
  enode_test_2();
  delete_create_get_set_test_1();
  create_get_stat_test();
}

@Override
public byte[] getData(String path, boolean watch, Stat stat)
        throws KeeperException, InterruptedException {
    int count = 0;
    do {
        try {
            return super.getData(path, watch ? watcher : null, stat);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return null;
}

/**
 * 初始化znode基础设置，生成3个永久的znode
 * /
 * |--- cobweb
 * |--- wvTasks
 * |--- wvWorkers
 * |--- wvManagers
 */
public void initZK() throws KeeperException, InterruptedException {
    try {
        createParent(ZNodeStaticSetting.WORKERS_PATH);
        createParent(ZNodeStaticSetting.MANAGERS_PATH);
        createParent(ZNodeStaticSetting.TASKS_PATH);
        createParent(ZNodeStaticSetting.FILTERS_ROOT);
    } catch (KeeperException.NodeExistsException e) {
        // pass is ok
    }
}

public void loadNode(ZkNode zkNode) throws KeeperException, InterruptedException {
    ZooKeeper zk = zookeeprClientFactory.createZookeeper();
    Stat stat = zk.exists(zkNode.getPath(), null);
    if(stat!=null){
        byte[] data = zk.getData(zkNode.getPath(), null, stat);
        zkNode.setContent(zkNode.parse(data));
    }
}

/**
 * A private method used to re-establish a zookeeper session with a peer cluster.
 * @param ke
 */
protected void reconnect(KeeperException ke) {
  if (ke instanceof ConnectionLossException || ke instanceof SessionExpiredException
      || ke instanceof AuthFailedException) {
    String clusterKey = ctx.getPeerConfig().getClusterKey();
    LOG.warn("Lost the ZooKeeper connection for peer " + clusterKey, ke);
    try {
      reloadZkWatcher();
    } catch (IOException io) {
      LOG.warn("Creation of ZookeeperWatcher failed for peer " + clusterKey, io);
    }
  }
}

private void lookForOrphans() {
  List<String> orphans;
  try {
    orphans = ZKUtil.listChildrenNoWatch(this.watcher, this.watcher.splitLogZNode);
    if (orphans == null) {
      LOG.warn("could not get children of " + this.watcher.splitLogZNode);
      return;
    }
  } catch (KeeperException e) {
    LOG.warn("could not get children of " + this.watcher.splitLogZNode + " "
        + StringUtils.stringifyException(e));
    return;
  }
  int rescan_nodes = 0;
  int listSize = orphans.size();
  for (int i = 0; i < listSize; i++) {
    String path = orphans.get(i);
    String nodepath = ZKUtil.joinZNode(watcher.splitLogZNode, path);
    if (ZKSplitLog.isRescanNode(watcher, nodepath)) {
      rescan_nodes++;
      LOG.debug("found orphan rescan node " + path);
    } else {
      LOG.info("found orphan task " + path);
    }
    getDataSetWatch(nodepath, zkretries);
  }
  LOG.info("Found " + (orphans.size() - rescan_nodes) + " orphan tasks and " + rescan_nodes
      + " rescan nodes");
}

@Test
public void testAuth() throws Exception {
    ZooKeeper zk = createClient();
    try {
        zk.create("/path1", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
        Assert.fail("Should have gotten exception.");
    } catch (KeeperException e) {
        // ok, exception as expected.
        LOG.info("Got exception as expected: " + e);
    }
    finally {
        zk.close();
    }
}

public void validAuth() throws Exception {
    ZooKeeper zk = createClient();
    // any multiple of 5 will do...
    zk.addAuthInfo("key", "25".getBytes());
    try {
        createNodePrintAcl(zk, "/valid", "testValidAuth");
        zk.getData("/abc", false, null);
        zk.setData("/abc", "testData3".getBytes(), -1);
    } catch (KeeperException.AuthFailedException e) {
        Assert.fail("test failed :" + e);
    } finally {
        zk.close();
    }
}

@SuppressWarnings("unchecked")
public List<ACL> getACL(String path, Stat stat)
        throws KeeperException.NoNodeException {
    DataNode n = nodes.get(path);
    if (n == null) {
        throw new KeeperException.NoNodeException();
    }
    synchronized (n) {
        n.copyStat(stat);
        return new ArrayList<ACL>(aclCache.convertLong(n.acl));
    }
}

/**
 * Utility function to ensure that the configured base znode exists.
 * This recursively creates the znode as well as all of its parents.
 */
public synchronized void ensureParentZNode()
    throws IOException, InterruptedException {
  Preconditions.checkState(!wantToBeInElection,
      "ensureParentZNode() may not be called while in the election");

  String pathParts[] = znodeWorkingDir.split("/");
  Preconditions.checkArgument(pathParts.length >= 1 &&
      pathParts[0].isEmpty(),
      "Invalid path: %s", znodeWorkingDir);

  StringBuilder sb = new StringBuilder();
  for (int i = 1; i < pathParts.length; i++) {
    sb.append("/").append(pathParts[i]);
    String prefixPath = sb.toString();
    LOG.debug("Ensuring existence of " + prefixPath);
    try {
      createWithRetries(prefixPath, new byte[]{}, zkAcl, CreateMode.PERSISTENT);
    } catch (KeeperException e) {
      if (isNodeExists(e.code())) {
        // This is OK - just ensuring existence.
        continue;
      } else {
        throw new IOException("Couldn't create " + prefixPath, e);
      }
    }
  }

  LOG.info("Successfully created " + znodeWorkingDir + " in ZK.");
}

@Test
public void testTrustedAuth() {
    X509AuthenticationProvider provider = createProvider(clientCert);
    MockServerCnxn cnxn = new MockServerCnxn();
    cnxn.clientChain = new X509Certificate[] { clientCert };
    Assert.assertEquals(KeeperException.Code.OK, provider.handleAuthentication(cnxn, null));
}