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

private void addLastSeqIdsToOps(String queueId, Map<String, Long> lastSeqIds,
    List<ZKUtilOp> listOfOps) throws KeeperException, ReplicationException {
  String peerId = new ReplicationQueueInfo(queueId).getPeerId();
  for (Entry<String, Long> lastSeqEntry : lastSeqIds.entrySet()) {
    String path = getSerialReplicationRegionPeerNode(lastSeqEntry.getKey(), peerId);
    Pair<Long, Integer> p = getLastSequenceIdWithVersion(lastSeqEntry.getKey(), peerId);
    byte[] data = ZKUtil.positionToByteArray(lastSeqEntry.getValue());
    if (p.getSecond() < 0) { // ZNode does not exist.
      ZKUtil.createWithParents(zookeeper,
        path.substring(0, path.lastIndexOf(ZNodePaths.ZNODE_PATH_SEPARATOR)));
      listOfOps.add(ZKUtilOp.createAndFailSilent(path, data));
      continue;
    }
    // Perform CAS in a specific version v0 (HBASE-20138)
    int v0 = p.getSecond();
    long lastPushedSeqId = p.getFirst();
    if (lastSeqEntry.getValue() <= lastPushedSeqId) {
      continue;
    }
    listOfOps.add(ZKUtilOp.setData(path, data, v0));
  }
}
 

@Override
public void setLastSequenceIds(String peerId, Map<String, Long> lastSeqIds)
    throws ReplicationException {
  try {
    // No need CAS and retry here, because it'll call setLastSequenceIds() for disabled peers
    // only, so no conflict happen.
    List<ZKUtilOp> listOfOps = new ArrayList<>();
    for (Entry<String, Long> lastSeqEntry : lastSeqIds.entrySet()) {
      String path = getSerialReplicationRegionPeerNode(lastSeqEntry.getKey(), peerId);
      ZKUtil.createWithParents(zookeeper, path);
      listOfOps.add(ZKUtilOp.setData(path, ZKUtil.positionToByteArray(lastSeqEntry.getValue())));
    }
    if (!listOfOps.isEmpty()) {
      ZKUtil.multiOrSequential(zookeeper, listOfOps, true);
    }
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to set last sequence ids, peerId=" + peerId
        + ", size of lastSeqIds=" + lastSeqIds.size(), e);
  }
}
 

@Override
public void addPeer(String peerId, ReplicationPeerConfig peerConfig, boolean enabled,
    SyncReplicationState syncReplicationState) throws ReplicationException {
  List<ZKUtilOp> multiOps = Arrays.asList(
    ZKUtilOp.createAndFailSilent(getPeerNode(peerId),
      ReplicationPeerConfigUtil.toByteArray(peerConfig)),
    ZKUtilOp.createAndFailSilent(getPeerStateNode(peerId),
      enabled ? ENABLED_ZNODE_BYTES : DISABLED_ZNODE_BYTES),
    ZKUtilOp.createAndFailSilent(getSyncReplicationStateNode(peerId),
      SyncReplicationState.toByteArray(syncReplicationState)),
    ZKUtilOp.createAndFailSilent(getNewSyncReplicationStateNode(peerId), NONE_STATE_ZNODE_BYTES));
  try {
    ZKUtil.createWithParents(zookeeper, peersZNode);
    ZKUtil.multiOrSequential(zookeeper, multiOps, false);
  } catch (KeeperException e) {
    throw new ReplicationException(
      "Could not add peer with id=" + peerId + ", peerConfig=>" + peerConfig + ", state=" +
        (enabled ? "ENABLED" : "DISABLED") + ", syncReplicationState=" + syncReplicationState,
      e);
  }
}
 

/**
 * Starts the syncer
 * @throws KeeperException if error occurs when trying to create base nodes on client ZK
 */
public void start() throws KeeperException {
  LOG.debug("Starting " + getClass().getSimpleName());
  this.watcher.registerListener(this);
  // create base znode on remote ZK
  ZKUtil.createWithParents(clientZkWatcher, watcher.getZNodePaths().baseZNode);
  // set meta znodes for client ZK
  Collection<String> nodes = getNodesToWatch();
  LOG.debug("Znodes to watch: " + nodes);
  // initialize queues and threads
  for (String node : nodes) {
    BlockingQueue<byte[]> queue = new ArrayBlockingQueue<>(1);
    queues.put(node, queue);
    Thread updater = new ClientZkUpdater(node, queue);
    updater.setDaemon(true);
    updater.start();
    watchAndCheckExists(node);
  }
}
 

/**
 * Top-level watcher/controller for procedures across the cluster.
 * <p>
 * On instantiation, this ensures the procedure znodes exist.  This however requires the passed in
 *  watcher has been started.
 * @param watcher watcher for the cluster ZK. Owned by <tt>this</tt> and closed via
 *          {@link #close()}
 * @param procDescription name of the znode describing the procedure to run
 * @throws KeeperException when the procedure znodes cannot be created
 */
public ZKProcedureUtil(ZKWatcher watcher, String procDescription)
    throws KeeperException {
  super(watcher);
  // make sure we are listening for events
  watcher.registerListener(this);
  // setup paths for the zknodes used in procedures
  this.baseZNode = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, procDescription);
  acquiredZnode = ZNodePaths.joinZNode(baseZNode, ACQUIRED_BARRIER_ZNODE_DEFAULT);
  reachedZnode = ZNodePaths.joinZNode(baseZNode, REACHED_BARRIER_ZNODE_DEFAULT);
  abortZnode = ZNodePaths.joinZNode(baseZNode, ABORT_ZNODE_DEFAULT);

  // first make sure all the ZK nodes exist
  // make sure all the parents exist (sometimes not the case in tests)
  ZKUtil.createWithParents(watcher, acquiredZnode);
  // regular create because all the parents exist
  ZKUtil.createAndFailSilent(watcher, reachedZnode);
  ZKUtil.createAndFailSilent(watcher, abortZnode);
}
 

/**
 * Setup the config for the cluster
 */
@BeforeClass
public static void setupCluster() throws Exception {
  setupConf(UTIL.getConfiguration());
  UTIL.startMiniZKCluster();
  UTIL.getConfiguration().setClass(MockRegistry.REGISTRY_IMPL_CONF_KEY, MockRegistry.class,
    DummyConnectionRegistry.class);
  CONNECTION = ConnectionFactory.createConnection(UTIL.getConfiguration());
  archivingClient = new ZKTableArchiveClient(UTIL.getConfiguration(), CONNECTION);
  // make hfile archiving node so we can archive files
  ZKWatcher watcher = UTIL.getZooKeeperWatcher();
  String archivingZNode = ZKTableArchiveClient.getArchiveZNode(UTIL.getConfiguration(), watcher);
  ZKUtil.createWithParents(watcher, archivingZNode);
  rss = mock(RegionServerServices.class);
  POOL = new DirScanPool(UTIL.getConfiguration());
}
 

public static boolean acquireLock(ZKWatcher zooKeeperWatcher, String parentNode,
        String lockName) throws KeeperException, InterruptedException {
    // Create the parent node as Persistent
    LOGGER.info("Creating the parent lock node:" + parentNode);
    ZKUtil.createWithParents(zooKeeperWatcher, parentNode);

    // Create the ephemeral node
    String lockNode = parentNode + "/" + lockName;
    String nodeValue = getHostName() + "_" + UUID.randomUUID().toString();
    LOGGER.info("Trying to acquire the lock by creating node:" + lockNode + " value:" + nodeValue);
    // Create the ephemeral node
    try {
        zooKeeperWatcher.getRecoverableZooKeeper().create(lockNode, Bytes.toBytes(nodeValue),
            Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    } catch (KeeperException.NodeExistsException e) {
        LOGGER.info("Could not acquire lock. Another process had already acquired the lock on Node "
                + lockName);
        return false;
    }
    LOGGER.info("Obtained the lock :" + lockNode);
    return true;
}
 

@Override
public void addWAL(ServerName serverName, String queueId, String fileName)
    throws ReplicationException {
  try {
    ZKUtil.createWithParents(zookeeper, getFileNode(serverName, queueId, fileName));
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to add wal to queue (serverName=" + serverName
        + ", queueId=" + queueId + ", fileName=" + fileName + ")", e);
  }
}
 

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

private ZKReplicationQueueStorage createWithUnstableVersion() throws IOException {
  return new ZKReplicationQueueStorage(UTIL.getZooKeeperWatcher(), UTIL.getConfiguration()) {

    private int called = 0;
    private int getLastSeqIdOpIndex = 0;

    @Override
    protected int getQueuesZNodeCversion() throws KeeperException {
      if (called < 4) {
        called++;
      }
      return called;
    }

    @Override
    protected Pair<Long, Integer> getLastSequenceIdWithVersion(String encodedRegionName,
        String peerId) throws KeeperException {
      Pair<Long, Integer> oldPair = super.getLastSequenceIdWithVersion(encodedRegionName, peerId);
      if (getLastSeqIdOpIndex < 100) {
        // Let the ZNode version increase.
        String path = getSerialReplicationRegionPeerNode(encodedRegionName, peerId);
        ZKUtil.createWithParents(zookeeper, path);
        ZKUtil.setData(zookeeper, path, ZKUtil.positionToByteArray(100L));
      }
      getLastSeqIdOpIndex++;
      return oldPair;
    }
  };
}
 

private static String initPeerClusterState(String baseZKNode)
    throws IOException, KeeperException {
  // Add a dummy region server and set up the cluster id
  Configuration testConf = new Configuration(conf);
  testConf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, baseZKNode);
  ZKWatcher zkw1 = new ZKWatcher(testConf, "test1", null);
  String fakeRs = ZNodePaths.joinZNode(zkw1.getZNodePaths().rsZNode,
          "hostname1.example.org:1234");
  ZKUtil.createWithParents(zkw1, fakeRs);
  ZKClusterId.setClusterId(zkw1, new ClusterId());
  return ZKConfig.getZooKeeperClusterKey(testConf);
}
 

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

public void start() throws KeeperException {
  watcher.registerListener(this);
  // make sure the base node exists
  ZKUtil.createWithParents(watcher, keysParentZNode);

  if (ZKUtil.watchAndCheckExists(watcher, keysParentZNode)) {
    List<ZKUtil.NodeAndData> nodes =
        ZKUtil.getChildDataAndWatchForNewChildren(watcher, keysParentZNode);
    refreshNodes(nodes);
  }
}
 

/***
 * Write a table's access controls to the permissions mirror in zookeeper
 * @param entry
 * @param permsData
 */
public void writeToZookeeper(byte[] entry, byte[] permsData) {
  String entryName = Bytes.toString(entry);
  String zkNode = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, ACL_NODE);
  zkNode = ZNodePaths.joinZNode(zkNode, entryName);

  try {
    ZKUtil.createWithParents(watcher, zkNode);
    ZKUtil.updateExistingNodeData(watcher, zkNode, permsData, -1);
  } catch (KeeperException e) {
    LOG.error("Failed updating permissions for entry '" +
        entryName + "'", e);
    watcher.abort("Failed writing node "+zkNode+" to zookeeper", e);
  }
}
 

@BeforeClass
public static void setUpBeforeClass() throws Exception {
  utility = new HBaseTestingUtility();
  utility.startMiniZKCluster();
  conf = utility.getConfiguration();
  ZKWatcher zk = HBaseTestingUtility.getZooKeeperWatcher(utility);
  ZKUtil.createWithParents(zk, zk.getZNodePaths().rsZNode);
}
 

@Test
public void testGetListOfRegionServers() throws Exception {
  // 0 region servers
  assertEquals(0, rt.getListOfRegionServers().size());

  // 1 region server
  ZKUtil.createWithParents(zkw,
    ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname1.example.org:1234"));
  List<String> rss = rt.getListOfRegionServers();
  assertEquals(rss.toString(), 1, rss.size());

  // 2 region servers
  ZKUtil.createWithParents(zkw,
    ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"));
  rss = rt.getListOfRegionServers();
  assertEquals(rss.toString(), 2, rss.size());

  // 1 region server
  ZKUtil.deleteNode(zkw,
    ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname2.example.org:1234"));
  rss = rt.getListOfRegionServers();
  assertEquals(1, rss.size());

  // 0 region server
  ZKUtil.deleteNode(zkw,
    ZNodePaths.joinZNode(zkw.getZNodePaths().rsZNode, "hostname1.example.org:1234"));
  rss = rt.getListOfRegionServers();
  assertEquals(rss.toString(), 0, rss.size());
}
 

@Before
public void setUp() throws Exception {
  Configuration conf = testUtil.getConfiguration();
  conf.set(HConstants.MASTER_PORT, "0");
  conf.setInt(HConstants.ZK_SESSION_TIMEOUT, 2000);
  testUtil.startMiniZKCluster();

  ZKWatcher watcher = testUtil.getZooKeeperWatcher();
  ZKUtil.createWithParents(watcher, watcher.getZNodePaths().masterAddressZNode,
          Bytes.toBytes("fake:123"));
  master = new HMaster(conf);
  rpcClient = RpcClientFactory.createClient(conf, HConstants.CLUSTER_ID_DEFAULT);
}
 

protected static void setupZkAndReplication() throws Exception {
  // The implementing class should set up the conf
  assertNotNull(conf);
  zkw = new ZKWatcher(conf, "test", null);
  ZKUtil.createWithParents(zkw, "/hbase/replication");
  ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1");
  ZKUtil.setData(zkw, "/hbase/replication/peers/1",
      Bytes.toBytes(conf.get(HConstants.ZOOKEEPER_QUORUM) + ":"
          + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) + ":/1"));
  ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1/peer-state");
  ZKUtil.setData(zkw, "/hbase/replication/peers/1/peer-state",
    ZKReplicationPeerStorage.ENABLED_ZNODE_BYTES);
  ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1/sync-rep-state");
  ZKUtil.setData(zkw, "/hbase/replication/peers/1/sync-rep-state",
    ZKReplicationPeerStorage.NONE_STATE_ZNODE_BYTES);
  ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1/new-sync-rep-state");
  ZKUtil.setData(zkw, "/hbase/replication/peers/1/new-sync-rep-state",
    ZKReplicationPeerStorage.NONE_STATE_ZNODE_BYTES);
  ZKUtil.createWithParents(zkw, "/hbase/replication/state");
  ZKUtil.setData(zkw, "/hbase/replication/state", ZKReplicationPeerStorage.ENABLED_ZNODE_BYTES);

  ZKClusterId.setClusterId(zkw, new ClusterId());
  CommonFSUtils.setRootDir(utility.getConfiguration(), utility.getDataTestDir());
  fs = FileSystem.get(conf);
  oldLogDir = utility.getDataTestDir(HConstants.HREGION_OLDLOGDIR_NAME);
  logDir = utility.getDataTestDir(HConstants.HREGION_LOGDIR_NAME);
  remoteLogDir = utility.getDataTestDir(ReplicationUtils.REMOTE_WAL_DIR_NAME);
  replication = new Replication();
  replication.initialize(new DummyServer(), fs, logDir, oldLogDir, null);
  managerOfCluster = getManagerFromCluster();
  if (managerOfCluster != null) {
    // After replication procedure, we need to add peer by hand (other than by receiving
    // notification from zk)
    managerOfCluster.addPeer(slaveId);
  }

  manager = replication.getReplicationManager();
  manager.addSource(slaveId);
  if (managerOfCluster != null) {
    waitPeer(slaveId, managerOfCluster, true);
  }
  waitPeer(slaveId, manager, true);

  htd = TableDescriptorBuilder.newBuilder(test)
    .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(f1)
      .setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build())
    .setColumnFamily(ColumnFamilyDescriptorBuilder.of(f2)).build();

  scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
  for(byte[] fam : htd.getColumnFamilyNames()) {
    scopes.put(fam, 0);
  }
  hri = RegionInfoBuilder.newBuilder(htd.getTableName()).setStartKey(r1).setEndKey(r2).build();
}