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

@Override
public int remainingTasksInCoordination() {
  int count = 0;
  try {
    List<String> tasks = ZKUtil.listChildrenNoWatch(watcher,
            watcher.getZNodePaths().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;
}
 

@Override
public void removeLastSequenceIds(String peerId) throws ReplicationException {
  String suffix = "-" + peerId;
  try {
    StringBuilder sb = new StringBuilder(regionsZNode);
    int regionsZNodeLength = regionsZNode.length();
    int levelOneLength = regionsZNodeLength + 3;
    int levelTwoLength = levelOneLength + 3;
    List<String> levelOneDirs = ZKUtil.listChildrenNoWatch(zookeeper, regionsZNode);
    // it is possible that levelOneDirs is null if we haven't write any last pushed sequence ids
    // yet, so we need an extra check here.
    if (CollectionUtils.isEmpty(levelOneDirs)) {
      return;
    }
    for (String levelOne : levelOneDirs) {
      sb.append(ZNodePaths.ZNODE_PATH_SEPARATOR).append(levelOne);
      for (String levelTwo : ZKUtil.listChildrenNoWatch(zookeeper, sb.toString())) {
        sb.append(ZNodePaths.ZNODE_PATH_SEPARATOR).append(levelTwo);
        for (String znode : ZKUtil.listChildrenNoWatch(zookeeper, sb.toString())) {
          if (znode.endsWith(suffix)) {
            sb.append(ZNodePaths.ZNODE_PATH_SEPARATOR).append(znode);
            ZKUtil.deleteNode(zookeeper, sb.toString());
            sb.setLength(levelTwoLength);
          }
        }
        sb.setLength(levelOneLength);
      }
      sb.setLength(regionsZNodeLength);
    }
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to remove all last sequence ids, peerId=" + peerId, e);
  }
}
 

private List<ServerName> getListOfReplicators0() throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(zookeeper, queuesZNode);
  if (children == null) {
    children = Collections.emptyList();
  }
  return children.stream().map(ServerName::parseServerName).collect(toList());
}
 

@Override
public List<String> listPeerIds() throws ReplicationException {
  try {
    List<String> children = ZKUtil.listChildrenNoWatch(zookeeper, peersZNode);
    return children != null ? children : Collections.emptyList();
  } catch (KeeperException e) {
    throw new ReplicationException("Cannot get the list of peers", e);
  }
}
 

/**
 * Get a list of all the other region servers in this cluster and set a watch
 * @return a list of server nanes
 */
private List<String> getRegisteredRegionServers(boolean watch) {
  List<String> result = null;
  try {
    if (watch) {
      result = ZKUtil.listChildrenAndWatchThem(this.zookeeper,
              this.zookeeper.getZNodePaths().rsZNode);
    } else {
      result = ZKUtil.listChildrenNoWatch(this.zookeeper, this.zookeeper.getZNodePaths().rsZNode);
    }
  } catch (KeeperException e) {
    this.abortable.abort("Get list of registered region servers", e);
  }
  return result;
}
 

private void lookForOrphans() {
  List<String> orphans;
  try {
    orphans = ZKUtil.listChildrenNoWatch(this.watcher,
            this.watcher.getZNodePaths().splitLogZNode);
    if (orphans == null) {
      LOG.warn("Could not get children of " + this.watcher.getZNodePaths().splitLogZNode);
      return;
    }
  } catch (KeeperException e) {
    LOG.warn("Could not get children of " + this.watcher.getZNodePaths().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 = ZNodePaths.joinZNode(watcher.getZNodePaths().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");
}
 

/**
 * Helper method to print the current state of the ZK tree.
 * @see #logZKTree(String)
 * @throws KeeperException if an unexpected exception occurs
 */
protected void logZKTree(String root, String prefix) throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(watcher, root);
  if (children == null) return;
  for (String child : children) {
    LOG.debug(prefix + child);
    String node = ZNodePaths.joinZNode(root.equals("/") ? "" : root, child);
    logZKTree(node, prefix + "---");
  }
}
 

/**
 * Method for table states migration.
 * Used when upgrading from pre-2.0 to 2.0
 * Reading state from zk, applying them to internal state
 * and delete.
 * Used by master to clean migration from zk based states to
 * table descriptor based states.
 * @deprecated Since 2.0.0. To be removed in hbase-3.0.0.
 */
@Deprecated
public static Map<TableName, TableState.State> queryForTableStates(ZKWatcher zkw)
    throws KeeperException, InterruptedException {
  Map<TableName, TableState.State> rv = new HashMap<>();
  List<String> children = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().tableZNode);
  if (children == null)
    return rv;
  for (String child: children) {
    TableName tableName = TableName.valueOf(child);
    ZooKeeperProtos.DeprecatedTableState.State state = getTableState(zkw, tableName);
    TableState.State newState = TableState.State.ENABLED;
    if (state != null) {
      switch (state) {
      case ENABLED:
        newState = TableState.State.ENABLED;
        break;
      case DISABLED:
        newState = TableState.State.DISABLED;
        break;
      case DISABLING:
        newState = TableState.State.DISABLING;
        break;
      case ENABLING:
        newState = TableState.State.ENABLING;
        break;
      default:
      }
    }
    rv.put(tableName, newState);
  }
  return rv;
}
 

/**
 * Tests the replication scenario 0 -> 0. By default
 * {@link BaseReplicationEndpoint#canReplicateToSameCluster()} returns false, so the
 * ReplicationSource should terminate, and no further logs should get enqueued
 */
@Test
public void testLoopedReplication() throws Exception {
  LOG.info("testLoopedReplication");
  startMiniClusters(1);
  createTableOnClusters(table);
  addPeer("1", 0, 0);
  Thread.sleep(SLEEP_TIME);

  // wait for source to terminate
  final ServerName rsName = utilities[0].getHBaseCluster().getRegionServer(0).getServerName();
  Waiter.waitFor(baseConfiguration, 10000, new Waiter.Predicate<Exception>() {
    @Override
    public boolean evaluate() throws Exception {
      ClusterMetrics clusterStatus = utilities[0].getAdmin()
          .getClusterMetrics(EnumSet.of(ClusterMetrics.Option.LIVE_SERVERS));
      ServerMetrics serverLoad = clusterStatus.getLiveServerMetrics().get(rsName);
      List<ReplicationLoadSource> replicationLoadSourceList =
          serverLoad.getReplicationLoadSourceList();
      return replicationLoadSourceList.isEmpty();
    }
  });

  Table[] htables = getHTablesOnClusters(tableName);
  putAndWait(row, famName, htables[0], htables[0]);
  rollWALAndWait(utilities[0], table.getTableName(), row);
  ZKWatcher zkw = utilities[0].getZooKeeperWatcher();
  String queuesZnode = ZNodePaths.joinZNode(zkw.getZNodePaths().baseZNode,
    ZNodePaths.joinZNode("replication", "rs"));
  List<String> listChildrenNoWatch =
      ZKUtil.listChildrenNoWatch(zkw, ZNodePaths.joinZNode(queuesZnode, rsName.toString()));
  assertEquals(0, listChildrenNoWatch.size());
}
 

@Test
public void testRegionServerHostname() throws Exception {
  Enumeration<NetworkInterface> netInterfaceList = NetworkInterface.getNetworkInterfaces();
  while (netInterfaceList.hasMoreElements()) {
    NetworkInterface ni = netInterfaceList.nextElement();
    Enumeration<InetAddress> addrList = ni.getInetAddresses();
    // iterate through host addresses and use each as hostname
    while (addrList.hasMoreElements()) {
      InetAddress addr = addrList.nextElement();
      if (addr.isLoopbackAddress() || addr.isLinkLocalAddress() || addr.isMulticastAddress() ||
          !addr.isSiteLocalAddress()) {
        continue;
      }
      String hostName = addr.getHostName();
      LOG.info("Found " + hostName + " on " + ni + ", addr=" + addr);

      TEST_UTIL.getConfiguration().set(DNS.MASTER_HOSTNAME_KEY, hostName);
      TEST_UTIL.getConfiguration().set(DNS.RS_HOSTNAME_KEY, hostName);
      StartMiniClusterOption option = StartMiniClusterOption.builder()
          .numMasters(NUM_MASTERS).numRegionServers(NUM_RS).numDataNodes(NUM_RS).build();
      TEST_UTIL.startMiniCluster(option);
      try {
        ZKWatcher zkw = TEST_UTIL.getZooKeeperWatcher();
        List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode);
        // there would be NUM_RS+1 children - one for the master
        assertTrue(servers.size() ==
          NUM_RS + (LoadBalancer.isTablesOnMaster(TEST_UTIL.getConfiguration())? 1: 0));
        for (String server : servers) {
          assertTrue("From zookeeper: " + server + " hostname: " + hostName,
            server.startsWith(hostName.toLowerCase(Locale.ROOT)+","));
        }
        zkw.close();
      } finally {
        TEST_UTIL.shutdownMiniCluster();
      }
    }
  }
}
 

@Test
public void testRegionServerHostnameReportedToMaster() throws Exception {
  TEST_UTIL.getConfiguration().setBoolean(HRegionServer.RS_HOSTNAME_DISABLE_MASTER_REVERSEDNS_KEY,
  true);
  StartMiniClusterOption option = StartMiniClusterOption.builder()
      .numMasters(NUM_MASTERS).numRegionServers(NUM_RS).numDataNodes(NUM_RS).build();
  TEST_UTIL.startMiniCluster(option);
  boolean tablesOnMaster = LoadBalancer.isTablesOnMaster(TEST_UTIL.getConfiguration());
  int expectedRS = NUM_RS + (tablesOnMaster? 1: 0);
  try (ZKWatcher zkw = TEST_UTIL.getZooKeeperWatcher()) {
    List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode);
    assertEquals(expectedRS, servers.size());
  }
}
 

private List<String> getWALsInQueue0(ServerName serverName, String queueId)
    throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(zookeeper, getQueueNode(serverName,
      queueId));
  return children != null ? children : Collections.emptyList();
}
 

private List<String> getAllQueues0(ServerName serverName) throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(zookeeper, getRsNode(serverName));
  return children != null ? children : Collections.emptyList();
}
 

private List<String> getAllPeersFromHFileRefsQueue0() throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(zookeeper, hfileRefsZNode);
  return children != null ? children : Collections.emptyList();
}
 

private List<String> getReplicableHFiles0(String peerId) throws KeeperException {
  List<String> children = ZKUtil.listChildrenNoWatch(this.zookeeper,
      getHFileRefsPeerNode(peerId));
  return children != null ? children : Collections.emptyList();
}
 

private List<String> getRegionServersInZK(final ZKWatcher zkw)
throws KeeperException {
  return ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().rsZNode);
}
 

@Test
public void testRescan() throws Exception {
  LOG.info("testRescan");
  SplitLogCounters.resetCounters();
  final ServerName SRV = ServerName.valueOf("svr,1,1");
  RegionServerServices mockedRS = getRegionServer(SRV);
  slw = new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
  slw.start();
  Thread.yield(); // let the worker start
  Thread.sleep(100);

  String task = ZKSplitLog.getEncodedNodeName(zkw, "task");
  SplitLogTask slt = new SplitLogTask.Unassigned(MANAGER);
  zkw.getRecoverableZooKeeper().create(task,slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
    CreateMode.PERSISTENT);

  waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
  // now the worker is busy doing the above task

  // preempt the task, have it owned by another worker
  ZKUtil.setData(zkw, task, slt.toByteArray());
  waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);

  // create a RESCAN node
  String rescan = ZKSplitLog.getEncodedNodeName(zkw, "RESCAN");
  rescan = zkw.getRecoverableZooKeeper().create(rescan, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
    CreateMode.PERSISTENT_SEQUENTIAL);

  waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 1, 2, WAIT_TIME);
  // RESCAN node might not have been processed if the worker became busy
  // with the above task. preempt the task again so that now the RESCAN
  // node is processed
  ZKUtil.setData(zkw, task, slt.toByteArray());
  waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 1, 2, WAIT_TIME);
  waitForCounter(SplitLogCounters.tot_wkr_task_acquired_rescan, 0, 1, WAIT_TIME);

  List<String> nodes = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().splitLogZNode);
  LOG.debug(Objects.toString(nodes));
  int num = 0;
  for (String node : nodes) {
    num++;
    if (node.startsWith("RESCAN")) {
      String name = ZKSplitLog.getEncodedNodeName(zkw, node);
      String fn = ZKSplitLog.getFileName(name);
      byte [] data = ZKUtil.getData(zkw,
              ZNodePaths.joinZNode(zkw.getZNodePaths().splitLogZNode, fn));
      slt = SplitLogTask.parseFrom(data);
      assertTrue(slt.toString(), slt.isDone(SRV));
    }
  }
  assertEquals(2, num);
}