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

@Override
public void nodeDataChanged(String path) {
  if (path.equals(labelZnode) || path.equals(userAuthsZnode)) {
    try {
      watcher.syncOrTimeout(path);
      byte[] data = ZKUtil.getDataAndWatch(watcher, path);
      if (path.equals(labelZnode)) {
        refreshVisibilityLabelsCache(data);
      } else {
        refreshUserAuthsCache(data);
      }
    } catch (KeeperException ke) {
      LOG.error("Error reading data from zookeeper for node " + path, ke);
      // only option is to abort
      watcher.abort("ZooKeeper error getting data for node " + path, ke);
    }
  }
}
 

@Override
public void nodeDataChanged(String path) {
  if (keysParentZNode.equals(ZKUtil.getParent(path))) {
    try {
      byte[] data = ZKUtil.getDataAndWatch(watcher, path);
      if (data == null || data.length == 0) {
        LOG.debug("Ignoring empty node "+path);
        return;
      }

      AuthenticationKey key = (AuthenticationKey)Writables.getWritable(data,
          new AuthenticationKey());
      secretManager.addKey(key);
    } catch (KeeperException ke) {
      LOG.error(HBaseMarkers.FATAL, "Error reading data from zookeeper", ke);
      watcher.abort("Error reading updated key znode "+path, ke);
    } catch (IOException ioe) {
      LOG.error(HBaseMarkers.FATAL, "Error reading key writables", ioe);
      watcher.abort("Error reading key writables from znode "+path, ioe);
    }
  }
}
 

private void watchAndCheckExists(String node) {
  try {
    if (ZKUtil.watchAndCheckExists(watcher, node)) {
      byte[] data = ZKUtil.getDataAndWatch(watcher, node);
      if (data != null) {
        // put the data into queue
        upsertQueue(node, data);
      } else {
        // It existed but now does not, should has been tracked by our watcher, ignore
        LOG.debug("Found no data from " + node);
        watchAndCheckExists(node);
      }
    } else {
      // cleanup stale ZNodes on client ZK to avoid invalid requests to server
      ZKUtil.deleteNodeFailSilent(clientZkWatcher, node);
    }
  } catch (KeeperException e) {
    server.abort("Unexpected exception during initialization, aborting", e);
  }
}
 

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

@Override
public void nodeCreated(String path) {
  if (!validate(path)) {
    return;
  }
  try {
    byte[] data = ZKUtil.getDataAndWatch(watcher, path);
    upsertQueue(path, data);
  } catch (KeeperException e) {
    LOG.warn("Unexpected exception handling nodeCreated event", e);
  }
}
 

/**
 * Gets the HRegionLocation for a given meta replica ID. Renews the watch on the znode for
 * future updates.
 * @param replicaId ReplicaID of the region.
 * @return HRegionLocation for the meta replica.
 * @throws KeeperException if there is any issue fetching/parsing the serialized data.
 */
private HRegionLocation getMetaRegionLocation(int replicaId)
    throws KeeperException {
  RegionState metaRegionState;
  try {
    byte[] data = ZKUtil.getDataAndWatch(watcher,
        watcher.getZNodePaths().getZNodeForReplica(replicaId));
    metaRegionState = ProtobufUtil.parseMetaRegionStateFrom(data, replicaId);
  } catch (DeserializationException e) {
    throw ZKUtil.convert(e);
  }
  return new HRegionLocation(metaRegionState.getRegion(), metaRegionState.getServerName());
}