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

@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 addHFileRefs(String peerId, List<Pair<Path, Path>> pairs)
    throws ReplicationException {
  String peerNode = getHFileRefsPeerNode(peerId);
  LOG.debug("Adding hfile references {} in queue {}", pairs, peerNode);
  List<ZKUtilOp> listOfOps = pairs.stream().map(p -> p.getSecond().getName())
      .map(n -> getHFileNode(peerNode, n))
      .map(f -> ZKUtilOp.createAndFailSilent(f, HConstants.EMPTY_BYTE_ARRAY)).collect(toList());
  LOG.debug("The multi list size for adding hfile references in zk for node {} is {}",
        peerNode, listOfOps.size());
  try {
    ZKUtil.multiOrSequential(this.zookeeper, listOfOps, true);
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to add hfile reference to peer " + peerId, 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);
  }
}
 

@Override
public void removeLastSequenceIds(String peerId, List<String> encodedRegionNames)
    throws ReplicationException {
  try {
    List<ZKUtilOp> listOfOps =
      encodedRegionNames.stream().map(n -> getSerialReplicationRegionPeerNode(n, peerId))
        .map(ZKUtilOp::deleteNodeFailSilent).collect(Collectors.toList());
    ZKUtil.multiOrSequential(zookeeper, listOfOps, true);
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to remove last sequence ids, peerId=" + peerId +
      ", encodedRegionNames.size=" + encodedRegionNames.size(), e);
  }
}
 

@Override
public void removeHFileRefs(String peerId, List<String> files) throws ReplicationException {
  String peerNode = getHFileRefsPeerNode(peerId);
  LOG.debug("Removing hfile references {} from queue {}", files, peerNode);

  List<ZKUtilOp> listOfOps = files.stream().map(n -> getHFileNode(peerNode, n))
      .map(ZKUtilOp::deleteNodeFailSilent).collect(toList());
  LOG.debug("The multi list size for removing hfile references in zk for node {} is {}",
      peerNode, listOfOps.size());
  try {
    ZKUtil.multiOrSequential(this.zookeeper, listOfOps, true);
  } catch (KeeperException e) {
    throw new ReplicationException("Failed to remove hfile reference from peer " + peerId, e);
  }
}
 

@Override
public void transitPeerSyncReplicationState(String peerId) throws ReplicationException {
  String newStateNode = getNewSyncReplicationStateNode(peerId);
  try {
    byte[] data = ZKUtil.getData(zookeeper, newStateNode);
    ZKUtil.multiOrSequential(zookeeper,
      Arrays.asList(ZKUtilOp.setData(newStateNode, NONE_STATE_ZNODE_BYTES),
        ZKUtilOp.setData(getSyncReplicationStateNode(peerId), data)),
      false);
  } catch (KeeperException | InterruptedException e) {
    throw new ReplicationException(
      "Error transiting sync replication state for peer with id=" + peerId, e);
  }
}