org.apache.hadoop.hbase.zookeeper.ZKUtil#checkExists ( )源码实例Demo

private SyncReplicationState getSyncReplicationState(String peerId, String path)
    throws ReplicationException {
  try {
    byte[] data = ZKUtil.getData(zookeeper, path);
    if (data == null || data.length == 0) {
      if (ZKUtil.checkExists(zookeeper, getPeerNode(peerId)) != -1) {
        // should be a peer from previous version, set the sync replication state for it.
        ZKUtil.createSetData(zookeeper, path, NONE_STATE_ZNODE_BYTES);
        return SyncReplicationState.NONE;
      } else {
        throw new ReplicationException(
          "Replication peer sync state shouldn't be empty, peerId=" + peerId);
      }
    }
    return SyncReplicationState.parseFrom(data);
  } catch (KeeperException | InterruptedException | IOException e) {
    throw new ReplicationException(
      "Error getting sync replication state of path " + path + " for peer with id=" + peerId, e);
  }
}
 

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

private void updateZooKeeper(TableState tableState) throws IOException {
  if (tableState == null) {
    return;
  }
  String znode = ZNodePaths.joinZNode(this.master.getZooKeeper().getZNodePaths().tableZNode,
    tableState.getTableName().getNameAsString());
  try {
    // Make sure znode exists.
    if (ZKUtil.checkExists(this.master.getZooKeeper(), znode) == -1) {
      ZKUtil.createAndFailSilent(this.master.getZooKeeper(), znode);
    }
    // Now set newState
    ZooKeeperProtos.DeprecatedTableState.Builder builder =
      ZooKeeperProtos.DeprecatedTableState.newBuilder();
    builder.setState(
      ZooKeeperProtos.DeprecatedTableState.State.valueOf(tableState.getState().toString()));
    byte[] data = ProtobufUtil.prependPBMagic(builder.build().toByteArray());
    ZKUtil.setData(this.master.getZooKeeper(), znode, data);
  } catch (KeeperException e) {
    // Only hbase1 clients suffer if this fails.
    LOG.warn("Failed setting table state to zookeeper mirrored for hbase-1.x clients", e);
  }
}
 

public SplitOrMergeTracker(ZKWatcher watcher, Configuration conf,
                           Abortable abortable) {
  try {
    if (ZKUtil.checkExists(watcher, watcher.getZNodePaths().switchZNode) < 0) {
      ZKUtil.createAndFailSilent(watcher, watcher.getZNodePaths().switchZNode);
    }
  } catch (KeeperException e) {
    throw new RuntimeException(e);
  }
  splitZnode = ZNodePaths.joinZNode(watcher.getZNodePaths().switchZNode,
    conf.get("zookeeper.znode.switch.split", "split"));
  mergeZnode = ZNodePaths.joinZNode(watcher.getZNodePaths().switchZNode,
    conf.get("zookeeper.znode.switch.merge", "merge"));
  splitStateTracker = new SwitchStateTracker(watcher, splitZnode, abortable);
  mergeStateTracker = new SwitchStateTracker(watcher, mergeZnode, abortable);
}
 

/**
 * 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(ZKWatcher zooKeeper, byte[] table) throws KeeperException {
  // ensure the latest state of the archive node is found
  zooKeeper.syncOrTimeout(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.syncOrTimeout(tableNode);

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

@Test
public void testTaskResigned() throws Exception {
  LOG.info("TestTaskResigned - resubmit task node once in RESIGNED state");
  assertEquals(0, tot_mgr_resubmit.sum());
  slm = new SplitLogManager(master, conf);
  assertEquals(0, tot_mgr_resubmit.sum());
  TaskBatch batch = new TaskBatch();
  String tasknode = submitTaskAndWait(batch, "foo/1");
  assertEquals(0, tot_mgr_resubmit.sum());
  final ServerName worker1 = ServerName.valueOf("worker1,1,1");
  assertEquals(0, tot_mgr_resubmit.sum());
  SplitLogTask slt = new SplitLogTask.Resigned(worker1);
  assertEquals(0, tot_mgr_resubmit.sum());
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  ZKUtil.checkExists(zkw, tasknode);
  // Could be small race here.
  if (tot_mgr_resubmit.sum() == 0) {
    waitForCounter(tot_mgr_resubmit, 0, 1, to/2);
  }
  assertEquals(1, tot_mgr_resubmit.sum());

  byte[] taskstate = ZKUtil.getData(zkw, tasknode);
  slt = SplitLogTask.parseFrom(taskstate);
  assertTrue(slt.isUnassigned(master.getServerName()));
}
 

@Override
public void addPeerToHFileRefs(String peerId) throws ReplicationException {
  String peerNode = getHFileRefsPeerNode(peerId);
  try {
    if (ZKUtil.checkExists(zookeeper, peerNode) == -1) {
      LOG.info("Adding peer {} to hfile reference queue.", peerId);
      ZKUtil.createWithParents(zookeeper, peerNode);
    }
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to add peer " + peerId + " to hfile reference queue.",
        e);
  }
}
 

@Override
public void removePeerFromHFileRefs(String peerId) throws ReplicationException {
  String peerNode = getHFileRefsPeerNode(peerId);
  try {
    if (ZKUtil.checkExists(zookeeper, peerNode) == -1) {
      LOG.debug("Peer {} not found in hfile reference queue.", peerNode);
    } else {
      LOG.info("Removing peer {} from hfile reference queue.", peerNode);
      ZKUtil.deleteNodeRecursively(zookeeper, peerNode);
    }
  } catch (KeeperException e) {
    throw new ReplicationException(
        "Failed to remove peer " + peerId + " from hfile reference queue.", e);
  }
}
 

/**
 * @return True if cluster has an active master.
 */
boolean hasActiveMaster() {
  try {
    if (ZKUtil.checkExists(watcher, watcher.getZNodePaths().masterAddressZNode) >= 0) {
      return true;
    }
  }
  catch (KeeperException ke) {
    LOG.info("Received an unexpected KeeperException when checking " +
        "isActiveMaster : "+ ke);
  }
  return false;
}
 

/**
 * Stop tracking a table. Ensures that the table doesn't exist, but if it does, it attempts to add
 * the table back via {@link #addAndReWatchTable(String)} - its a 'safe' removal.
 * @param tableZnode full zookeeper path to the table to be added
 * @throws KeeperException if an unexpected zk exception occurs
 */
private void safeStopTrackingTable(String tableZnode) throws KeeperException {
  getMonitor().removeTable(ZKUtil.getNodeName(tableZnode));
  // if the table exists, then add and rewatch it
  if (ZKUtil.checkExists(watcher, tableZnode) >= 0) {
    addAndReWatchTable(tableZnode);
  }
}
 

private List<RSGroupInfo> retrieveGroupListFromZookeeper() throws IOException {
  String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
  List<RSGroupInfo> RSGroupInfoList = Lists.newArrayList();
  // Overwrite any info stored by table, this takes precedence
  try {
    if (ZKUtil.checkExists(watcher, groupBasePath) != -1) {
      List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(watcher, groupBasePath);
      if (children == null) {
        return RSGroupInfoList;
      }
      for (String znode : children) {
        byte[] data = ZKUtil.getData(watcher, ZNodePaths.joinZNode(groupBasePath, znode));
        if (data.length > 0) {
          ProtobufUtil.expectPBMagicPrefix(data);
          ByteArrayInputStream bis =
            new ByteArrayInputStream(data, ProtobufUtil.lengthOfPBMagic(), data.length);
          RSGroupInfoList
            .add(ProtobufUtil.toGroupInfo(RSGroupProtos.RSGroupInfo.parseFrom(bis)));
        }
      }
      LOG.debug("Read ZK GroupInfo count:" + RSGroupInfoList.size());
    }
  } catch (KeeperException | DeserializationException | InterruptedException e) {
    throw new IOException("Failed to read rsGroupZNode", e);
  }
  return RSGroupInfoList;
}
 

@Test
public void testUnassignedOrphan() throws Exception {
  LOG.info("TestUnassignedOrphan - an unassigned task is resubmitted at" +
      " startup");
  String tasknode = ZKSplitLog.getEncodedNodeName(zkw, "orphan/test/slash");
  //create an unassigned orphan task
  SplitLogTask slt = new SplitLogTask.Unassigned(master.getServerName());
  zkw.getRecoverableZooKeeper().create(tasknode, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
      CreateMode.PERSISTENT);
  int version = ZKUtil.checkExists(zkw, tasknode);

  slm = new SplitLogManager(master, conf);
  waitForCounter(tot_mgr_orphan_task_acquired, 0, 1, to/2);
  Task task = findOrCreateOrphanTask(tasknode);
  assertTrue(task.isOrphan());
  assertTrue(task.isUnassigned());
  // wait for RESCAN node to be created
  waitForCounter(tot_mgr_rescan, 0, 1, to / 2);
  Task task2 = findOrCreateOrphanTask(tasknode);
  assertTrue(task == task2);
  LOG.debug("task = " + task);
  assertEquals(1L, tot_mgr_resubmit.sum());
  assertEquals(1, task.incarnation.get());
  assertEquals(0, task.unforcedResubmits.get());
  assertTrue(task.isOrphan());
  assertTrue(task.isUnassigned());
  assertTrue(ZKUtil.checkExists(zkw, tasknode) > version);
}
 

@Test
public void testMultipleResubmits() throws Exception {
  LOG.info("TestMultipleResbmits - no indefinite resubmissions");
  conf.setInt("hbase.splitlog.max.resubmit", 2);
  slm = new SplitLogManager(master, conf);
  TaskBatch batch = new TaskBatch();

  String tasknode = submitTaskAndWait(batch, "foo/1");
  int version = ZKUtil.checkExists(zkw, tasknode);
  final ServerName worker1 = ServerName.valueOf("worker1,1,1");
  final ServerName worker2 = ServerName.valueOf("worker2,1,1");
  final ServerName worker3 = ServerName.valueOf("worker3,1,1");
  SplitLogTask slt = new SplitLogTask.Owned(worker1);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  waitForCounter(tot_mgr_heartbeat, 0, 1, to/2);
  waitForCounter(tot_mgr_resubmit, 0, 1, to + to/2);
  int version1 = ZKUtil.checkExists(zkw, tasknode);
  assertTrue(version1 > version);
  slt = new SplitLogTask.Owned(worker2);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  waitForCounter(tot_mgr_heartbeat, 1, 2, to/2);
  waitForCounter(tot_mgr_resubmit, 1, 2, to + to/2);
  int version2 = ZKUtil.checkExists(zkw, tasknode);
  assertTrue(version2 > version1);
  slt = new SplitLogTask.Owned(worker3);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  waitForCounter(tot_mgr_heartbeat, 2, 3, to/2);
  waitForCounter(tot_mgr_resubmit_threshold_reached, 0, 1, to + to/2);
  Thread.sleep(to + to/2);
  assertEquals(2L, tot_mgr_resubmit.sum() - tot_mgr_resubmit_force.sum());
}
 

@Test
public void testRescanCleanup() throws Exception {
  LOG.info("TestRescanCleanup - ensure RESCAN nodes are cleaned up");

  slm = new SplitLogManager(master, conf);
  TaskBatch batch = new TaskBatch();

  String tasknode = submitTaskAndWait(batch, "foo/1");
  int version = ZKUtil.checkExists(zkw, tasknode);
  final ServerName worker1 = ServerName.valueOf("worker1,1,1");
  SplitLogTask slt = new SplitLogTask.Owned(worker1);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  waitForCounter(tot_mgr_heartbeat, 0, 1, to/2);
  waitForCounter(new Expr() {
    @Override
    public long eval() {
      return (tot_mgr_resubmit.sum() + tot_mgr_resubmit_failed.sum());
    }
  }, 0, 1, 5*60000); // wait long enough
  Assert.assertEquals("Could not run test. Lost ZK connection?",
    0, tot_mgr_resubmit_failed.sum());
  int version1 = ZKUtil.checkExists(zkw, tasknode);
  assertTrue(version1 > version);
  byte[] taskstate = ZKUtil.getData(zkw, tasknode);
  slt = SplitLogTask.parseFrom(taskstate);
  assertTrue(slt.isUnassigned(master.getServerName()));

  waitForCounter(tot_mgr_rescan_deleted, 0, 1, to/2);
}
 

@Test
public void testDeadWorker() throws Exception {
  LOG.info("testDeadWorker");

  conf.setLong("hbase.splitlog.max.resubmit", 0);
  slm = new SplitLogManager(master, conf);
  TaskBatch batch = new TaskBatch();

  String tasknode = submitTaskAndWait(batch, "foo/1");
  int version = ZKUtil.checkExists(zkw, tasknode);
  final ServerName worker1 = ServerName.valueOf("worker1,1,1");
  SplitLogTask slt = new SplitLogTask.Owned(worker1);
  ZKUtil.setData(zkw, tasknode, slt.toByteArray());
  if (tot_mgr_heartbeat.sum() == 0) {
    waitForCounter(tot_mgr_heartbeat, 0, 1, to/2);
  }
  slm.handleDeadWorker(worker1);
  if (tot_mgr_resubmit.sum() == 0) {
    waitForCounter(tot_mgr_resubmit, 0, 1, to+to/2);
  }
  if (tot_mgr_resubmit_dead_server_task.sum() == 0) {
    waitForCounter(tot_mgr_resubmit_dead_server_task, 0, 1, to + to/2);
  }

  int version1 = ZKUtil.checkExists(zkw, tasknode);
  assertTrue(version1 > version);
  byte[] taskstate = ZKUtil.getData(zkw, tasknode);
  slt = SplitLogTask.parseFrom(taskstate);
  assertTrue(slt.isUnassigned(master.getServerName()));
  return;
}
 

private static void waitUntilZnodeAvailable(int replicaId) throws Exception {
  String znode = util.getZooKeeperWatcher().getZNodePaths().getZNodeForReplica(replicaId);
  int i = 0;
  while (i < 1000) {
    if (ZKUtil.checkExists(util.getZooKeeperWatcher(), znode) == -1) {
      Thread.sleep(100);
      i++;
    } else break;
  }
  if (i == 1000) throw new IOException("znode for meta replica " + replicaId + " not available");
}
 

/**
 * Check to see if the table is currently marked for archiving
 * @param table name of the table to check
 * @return <tt>true</tt> if the archive znode for that table exists, <tt>false</tt> if not
 * @throws KeeperException if an unexpected zookeeper error occurs
 */
public boolean isArchivingEnabled(byte[] table) throws KeeperException {
  String tableNode = this.getTableNode(table);
  return ZKUtil.checkExists(zooKeeper, tableNode) >= 0;
}