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

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.error(HBaseMarkers.FATAL, "Failed reading new secret key for id '" +
          keyId + "' from zk", ioe);
      watcher.abort("Error deserializing key from znode "+path, ioe);
    }
  }
}
 

private void refreshNodes(List<ZKUtil.NodeAndData> nodes) {
  for (ZKUtil.NodeAndData n : nodes) {
    if (Thread.interrupted()) {
      // Use Thread.interrupted so that we clear interrupt status
      break;
    }
    if (n.isEmpty()) continue;
    String path = n.getNode();
    String entry = (ZKUtil.getNodeName(path));
    try {
      refreshAuthManager(entry, n.getData());
    } catch (IOException ioe) {
      LOG.error("Failed parsing permissions for table '" + entry +
          "' from zk", ioe);
    }
  }
}
 

@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);
      LOG.info("Node deleted id={}", id);
    } catch (NumberFormatException nfe) {
      LOG.error("Invalid znode name for key ID '"+keyId+"'", nfe);
    }
  }
}
 

/**
 * Pass along the procedure global barrier notification to any listeners
 * @param path full znode path that cause the notification
 */
private void receivedReachedGlobalBarrier(String path) {
  LOG.debug("Received reached global barrier:" + path);
  String procName = ZKUtil.getNodeName(path);
  this.member.receivedReachedGlobalBarrier(procName);
}
 

/**
 * Start monitoring znodes in ZK - subclass hook to start monitoring znodes they are about.
 * @return true if succeed, false if encountered initialization errors.
 */
@Override
final public boolean start(final ProcedureCoordinator coordinator) {
  if (this.coordinator != null) {
    throw new IllegalStateException(
      "ZKProcedureCoordinator already started and already has listener installed");
  }
  this.coordinator = coordinator;

  try {
    this.zkProc = new ZKProcedureUtil(watcher, procedureType) {
      @Override
      public void nodeCreated(String path) {
        if (!isInProcedurePath(path)) return;
        LOG.debug("Node created: " + path);
        logZKTree(this.baseZNode);
        if (isAcquiredPathNode(path)) {
          // node wasn't present when we created the watch so zk event triggers acquire
          coordinator.memberAcquiredBarrier(ZKUtil.getNodeName(ZKUtil.getParent(path)),
            ZKUtil.getNodeName(path));
        } else if (isReachedPathNode(path)) {
          // node was absent when we created the watch so zk event triggers the finished barrier.

          // TODO Nothing enforces that acquire and reached znodes from showing up in wrong order.
          String procName = ZKUtil.getNodeName(ZKUtil.getParent(path));
          String member = ZKUtil.getNodeName(path);
          // get the data from the procedure member
          try {
            byte[] dataFromMember = ZKUtil.getData(watcher, path);
            // ProtobufUtil.isPBMagicPrefix will check null
            if (dataFromMember != null && dataFromMember.length > 0) {
              if (!ProtobufUtil.isPBMagicPrefix(dataFromMember)) {
                ForeignException ee = new ForeignException(coordName,
                  "Failed to get data from finished node or data is illegally formatted:"
                      + path);
                coordinator.abortProcedure(procName, ee);
              } else {
                dataFromMember = Arrays.copyOfRange(dataFromMember, ProtobufUtil.lengthOfPBMagic(),
                  dataFromMember.length);
                LOG.debug("Finished data from procedure '{}' member '{}': {}", procName, member,
                    new String(dataFromMember, StandardCharsets.UTF_8));
                coordinator.memberFinishedBarrier(procName, member, dataFromMember);
              }
            } else {
              coordinator.memberFinishedBarrier(procName, member, dataFromMember);
            }
          } catch (KeeperException e) {
            ForeignException ee = new ForeignException(coordName, e);
            coordinator.abortProcedure(procName, ee);
          } catch (InterruptedException e) {
            ForeignException ee = new ForeignException(coordName, e);
            coordinator.abortProcedure(procName, ee);
          }
        } else if (isAbortPathNode(path)) {
          abort(path);
        } else {
          LOG.debug("Ignoring created notification for node:" + path);
        }
      }
    };
    zkProc.clearChildZNodes();
  } catch (KeeperException e) {
    LOG.error("Unable to start the ZK-based Procedure Coordinator rpcs.", e);
    return false;
  }

  LOG.debug("Starting controller for procedure member=" + coordName);
  return true;
}