本文整理了Java中org.apache.hadoop.hbase.procedure.ZKProcedureMemberRpcs类的一些代码示例，展示了ZKProcedureMemberRpcs类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZKProcedureMemberRpcs类的具体详情如下：
包路径：org.apache.hadoop.hbase.procedure.ZKProcedureMemberRpcs
类名称：ZKProcedureMemberRpcs

ZKProcedureMemberRpcs介绍

[英]ZooKeeper based controller for a procedure member.

There can only be one ZKProcedureMemberRpcs per procedure type per member, since each procedure type is bound to a single set of znodes. You can have multiple ZKProcedureMemberRpcs on the same server, each serving a different member name, but each individual rpcs is still bound to a single member name (and since they are used to determine global progress, its important to not get this wrong).

To make this slightly more confusing, you can run multiple, concurrent procedures at the same time (as long as they have different types), from the same controller, but the same node name must be used for each procedure (though there is no conflict between the two procedure as long as they have distinct names).

There is no real error recovery with this mechanism currently -- if any the coordinator fails, its re-initialization will delete the znodes and require all in progress subprocedures to start anew.
[中]基于ZooKeeper的程序成员控制器。
每个成员的每个过程类型只能有一个ZKProcedurememberRPC，因为每个过程类型都绑定到一组ZNode。同一台服务器上可以有多个ZKProcedurememberPC，每个都提供不同的成员名称，但每个单独的RPC仍然绑定到一个成员名称（因为它们用于确定全局进度，所以不要弄错这一点很重要）。
为了让这一点稍微更令人困惑，您可以从同一个控制器同时运行多个并发过程（只要它们具有不同的类型），但每个过程都必须使用相同的节点名称（尽管两个过程之间没有冲突，只要它们具有不同的名称）。
目前这种机制没有真正的错误恢复——如果协调器出现任何故障，它的重新初始化将删除znode，并要求所有正在进行的子进程重新启动。

代码示例

代码示例来源：origin: apache/hbase

@Override
 public Pair<ZKProcedureCoordinator, List<ZKProcedureMemberRpcs>> start(
     ZKWatcher watcher, String operationName,
     ProcedureCoordinator coordinator, String controllerName,
     ProcedureMember member, List<String> expected) throws Exception {
  // start the controller
  ZKProcedureCoordinator controller = new ZKProcedureCoordinator(
    watcher, operationName, CONTROLLER_NODE_NAME);
  controller.start(coordinator);
  // make a cohort controller for each expected node
  List<ZKProcedureMemberRpcs> cohortControllers = new ArrayList<>();
  for (String nodeName : expected) {
   ZKProcedureMemberRpcs cc = new ZKProcedureMemberRpcs(watcher, operationName);
   cc.start(nodeName, member);
   cohortControllers.add(cc);
  }
  return new Pair<>(controller, cohortControllers);
 }
};

代码示例来源：origin: apache/hbase

private void watchForAbortedProcedures() {
 LOG.debug("Checking for aborted procedures on node: '" + zkController.getAbortZnode() + "'");
 try {
  // this is the list of the currently aborted procedues
  List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(zkController.getWatcher(),
         zkController.getAbortZnode());
  if (children == null || children.isEmpty()) {
   return;
  }
  for (String node : children) {
   String abortNode = ZNodePaths.joinZNode(zkController.getAbortZnode(), node);
   abort(abortNode);
  }
 } catch (KeeperException e) {
  member.controllerConnectionFailure("Failed to list children for abort node:"
    + zkController.getAbortZnode(), e, null);
 }
}

代码示例来源：origin: apache/hbase

} catch (IllegalArgumentException iae ) {
 LOG.error("Illegal argument exception", iae);
 sendMemberAborted(subproc, new ForeignException(getMemberName(), iae));
} catch (IllegalStateException ise) {
 LOG.error("Illegal state exception ", ise);
 sendMemberAborted(subproc, new ForeignException(getMemberName(), ise));
} catch (KeeperException e) {
 member.controllerConnectionFailure("Failed to get data for new procedure:" + opName,

代码示例来源：origin: apache/hbase

@Override
public void start(final String memberName, final ProcedureMember listener) {
 LOG.debug("Starting procedure member '" + memberName + "'");
 this.member = listener;
 this.memberName = memberName;
 watchForAbortedProcedures();
 waitForNewProcedures();
}

代码示例来源：origin: apache/hbase

@Override
 public ProcedureMemberRpcs getProcedureMemberRpcs(String procType) throws KeeperException {
  return new ZKProcedureMemberRpcs(watcher, procType);
 }
}

代码示例来源：origin: apache/hbase

final ZKProcedureMemberRpcs controller = new ZKProcedureMemberRpcs(
  watcher, "testSimple");
controller.start(COHORT_NODE_NAME, member);
String prepare = ZKProcedureUtil.getAcquireBarrierNode(controller.getZkController(), operationName);
ZKUtil.createSetData(watcher, prepare, ProtobufUtil.prependPBMagic(data));
String commit = ZKProcedureUtil.getReachedBarrierNode(controller.getZkController(), operationName);
LOG.debug("Found prepared, posting commit node:" + commit);
ZKUtil.createAndFailSilent(watcher, commit);
ZKUtil.deleteNodeRecursively(watcher, controller.getZkController().getBaseZnode());
assertEquals("Didn't delete prepare node", -1, ZKUtil.checkExists(watcher, prepare));
assertEquals("Didn't delete commit node", -1, ZKUtil.checkExists(watcher, commit));

代码示例来源：origin: apache/hbase

private void waitForNewProcedures() {
 // watch for new procedues that we need to start subprocedures for
 LOG.debug("Looking for new procedures under znode:'" + zkController.getAcquiredBarrier() + "'");
 List<String> runningProcedures = null;
 try {
  runningProcedures = ZKUtil.listChildrenAndWatchForNewChildren(zkController.getWatcher(),
   zkController.getAcquiredBarrier());
  if (runningProcedures == null) {
   LOG.debug("No running procedures.");
   return;
  }
 } catch (KeeperException e) {
  member.controllerConnectionFailure("General failure when watching for new procedures",
   e, null);
 }
 if (runningProcedures == null) {
  LOG.debug("No running procedures.");
  return;
 }
 for (String procName : runningProcedures) {
  // then read in the procedure information
  String path = ZNodePaths.joinZNode(zkController.getAcquiredBarrier(), procName);
  startNewSubprocedure(path);
 }
}

代码示例来源：origin: apache/hbase

/**
 * This attempts to create an acquired state znode for the procedure (snapshot name).
 *
 * It then looks for the reached znode to trigger in-barrier execution.  If not present we
 * have a watcher, if present then trigger the in-barrier action.
 */
@Override
public void sendMemberAcquired(Subprocedure sub) throws IOException {
 String procName = sub.getName();
 try {
  LOG.debug("Member: '" + memberName + "' joining acquired barrier for procedure (" + procName
    + ") in zk");
  String acquiredZNode = ZNodePaths.joinZNode(ZKProcedureUtil.getAcquireBarrierNode(
   zkController, procName), memberName);
  ZKUtil.createAndFailSilent(zkController.getWatcher(), acquiredZNode);
  // watch for the complete node for this snapshot
  String reachedBarrier = zkController.getReachedBarrierNode(procName);
  LOG.debug("Watch for global barrier reached:" + reachedBarrier);
  if (ZKUtil.watchAndCheckExists(zkController.getWatcher(), reachedBarrier)) {
   receivedReachedGlobalBarrier(reachedBarrier);
  }
 } catch (KeeperException e) {
  member.controllerConnectionFailure("Failed to acquire barrier for procedure: "
    + procName + " and member: " + memberName, e, procName);
 }
}

代码示例来源：origin: apache/hbase

LOG.error(msg);
 ee = new ForeignException(getMemberName(), new IllegalArgumentException(msg));
} else {
 data = Arrays.copyOfRange(data, ProtobufUtil.lengthOfPBMagic(), data.length);
  + " but we can't read the information. Killing the procedure.");
ee = new ForeignException(getMemberName(), e);

代码示例来源：origin: apache/hbase

/**
 * Initialize this region server flush procedure manager
 * Uses a zookeeper based member controller.
 * @param rss region server
 * @throws KeeperException if the zookeeper cannot be reached
 */
@Override
public void initialize(RegionServerServices rss) throws KeeperException {
 this.rss = rss;
 ZKWatcher zkw = rss.getZooKeeper();
 this.memberRpcs = new ZKProcedureMemberRpcs(zkw,
  MasterFlushTableProcedureManager.FLUSH_TABLE_PROCEDURE_SIGNATURE);
 Configuration conf = rss.getConfiguration();
 long keepAlive = conf.getLong(FLUSH_TIMEOUT_MILLIS_KEY, FLUSH_TIMEOUT_MILLIS_DEFAULT);
 int opThreads = conf.getInt(FLUSH_REQUEST_THREADS_KEY, FLUSH_REQUEST_THREADS_DEFAULT);
 // create the actual flush table procedure member
 ThreadPoolExecutor pool = ProcedureMember.defaultPool(rss.getServerName().toString(),
  opThreads, keepAlive);
 this.member = new ProcedureMember(memberRpcs, pool, new FlushTableSubprocedureBuilder());
}

代码示例来源：origin: co.cask.hbase/hbase

public void start(ProcedureMember listener) {
 LOG.debug("Starting procedure member '" + this.memberName + "'");
 this.member = listener;
 watchForAbortedProcedures();
 waitForNewProcedures();
}

代码示例来源：origin: org.apache.hbase/hbase-server

final ZKProcedureMemberRpcs controller = new ZKProcedureMemberRpcs(
  watcher, "testSimple");
controller.start(COHORT_NODE_NAME, member);
String prepare = ZKProcedureUtil.getAcquireBarrierNode(controller.getZkController(), operationName);
ZKUtil.createSetData(watcher, prepare, ProtobufUtil.prependPBMagic(data));
String commit = ZKProcedureUtil.getReachedBarrierNode(controller.getZkController(), operationName);
LOG.debug("Found prepared, posting commit node:" + commit);
ZKUtil.createAndFailSilent(watcher, commit);
ZKUtil.deleteNodeRecursively(watcher, controller.getZkController().getBaseZnode());
assertEquals("Didn't delete prepare node", -1, ZKUtil.checkExists(watcher, prepare));
assertEquals("Didn't delete commit node", -1, ZKUtil.checkExists(watcher, commit));

代码示例来源：origin: co.cask.hbase/hbase

private void waitForNewProcedures() {
 // watch for new procedues that we need to start subprocedures for
 LOG.debug("Looking for new procedures under znode:'" + zkController.getAcquiredBarrier() + "'");
 List<String> runningProcedures = null;
 try {
  runningProcedures = ZKUtil.listChildrenAndWatchForNewChildren(zkController.getWatcher(),
   zkController.getAcquiredBarrier());
  if (runningProcedures == null) {
   LOG.debug("No running procedures.");
   return;
  }
 } catch (KeeperException e) {
  member.controllerConnectionFailure("General failure when watching for new procedures",
   new IOException(e));
 }
 if (runningProcedures == null) {
  LOG.debug("No running procedures.");
  return;
 }
 for (String procName : runningProcedures) {
  // then read in the procedure information
  String path = ZKUtil.joinZNode(zkController.getAcquiredBarrier(), procName);
  startNewSubprocedure(path);
 }
}

代码示例来源：origin: co.cask.hbase/hbase

/**
 * This attempts to create an acquired state znode for the procedure (snapshot name).
 *
 * It then looks for the reached znode to trigger in-barrier execution.  If not present we
 * have a watcher, if present then trigger the in-barrier action.
 */
@Override
public void sendMemberAcquired(Subprocedure sub) throws IOException {
 String procName = sub.getName();
 try {
  LOG.debug("Member: '" + memberName + "' joining acquired barrier for procedure (" + procName
    + ") in zk");
  String acquiredZNode = ZKUtil.joinZNode(ZKProcedureUtil.getAcquireBarrierNode(
   zkController, procName), memberName);
  ZKUtil.createAndFailSilent(zkController.getWatcher(), acquiredZNode);
  // watch for the complete node for this snapshot
  String reachedBarrier = zkController.getReachedBarrierNode(procName);
  LOG.debug("Watch for global barrier reached:" + reachedBarrier);
  if (ZKUtil.watchAndCheckExists(zkController.getWatcher(), reachedBarrier)) {
   receivedReachedGlobalBarrier(reachedBarrier);
  }
 } catch (KeeperException e) {
  member.controllerConnectionFailure("Failed to acquire barrier for procedure: "
    + procName + " and member: " + memberName, new IOException(e));
 }
}

代码示例来源：origin: co.cask.hbase/hbase

LOG.error(msg);
 ee = new ForeignException(getMemberName(), new IllegalArgumentException(msg));
} else {
 data = Arrays.copyOfRange(data, ProtobufUtil.lengthOfPBMagic(), data.length);
  + " but we can't read the information. Killing the procedure.");
ee = new ForeignException(getMemberName(), e);

代码示例来源：origin: apache/hbase

@Override
 public Pair<ZKProcedureCoordinator, List<ZKProcedureMemberRpcs>> start(
     ZKWatcher watcher, String operationName,
     ProcedureCoordinator coordinator, String controllerName,
     ProcedureMember member, List<String> expected) throws Exception {
  // make a cohort controller for each expected node
  List<ZKProcedureMemberRpcs> cohortControllers = new ArrayList<>();
  for (String nodeName : expected) {
   ZKProcedureMemberRpcs cc = new ZKProcedureMemberRpcs(watcher, operationName);
   cc.start(nodeName, member);
   cohortControllers.add(cc);
  }
  // start the controller
  ZKProcedureCoordinator controller = new ZKProcedureCoordinator(
    watcher, operationName, CONTROLLER_NODE_NAME);
  controller.start(coordinator);
  return new Pair<>(controller, cohortControllers);
 }
};

代码示例来源：origin: apache/hbase

/**
 * Create a default snapshot handler - uses a zookeeper based member controller.
 * @param rss region server running the handler
 * @throws KeeperException if the zookeeper cluster cannot be reached
 */
@Override
public void initialize(RegionServerServices rss) throws KeeperException {
 this.rss = rss;
 ZKWatcher zkw = rss.getZooKeeper();
 this.memberRpcs = new ZKProcedureMemberRpcs(zkw,
   SnapshotManager.ONLINE_SNAPSHOT_CONTROLLER_DESCRIPTION);
 // read in the snapshot request configuration properties
 Configuration conf = rss.getConfiguration();
 long keepAlive = conf.getLong(SNAPSHOT_TIMEOUT_MILLIS_KEY, SNAPSHOT_TIMEOUT_MILLIS_DEFAULT);
 int opThreads = conf.getInt(SNAPSHOT_REQUEST_THREADS_KEY, SNAPSHOT_REQUEST_THREADS_DEFAULT);
 // create the actual snapshot procedure member
 ThreadPoolExecutor pool = ProcedureMember.defaultPool(rss.getServerName().toString(),
  opThreads, keepAlive);
 this.member = new ProcedureMember(memberRpcs, pool, new SnapshotSubprocedureBuilder());
}

代码示例来源：origin: harbby/presto-connectors

public void start(final String memberName, final ProcedureMember listener) {
 LOG.debug("Starting procedure member '" + memberName + "'");
 this.member = listener;
 this.memberName = memberName;
 watchForAbortedProcedures();
 waitForNewProcedures();
}

代码示例来源：origin: co.cask.hbase/hbase

} catch (IllegalArgumentException iae ) {
 LOG.error("Illegal argument exception", iae);
 sendMemberAborted(subproc, new ForeignException(getMemberName(), iae));
} catch (IllegalStateException ise) {
 LOG.error("Illegal state exception ", ise);
 sendMemberAborted(subproc, new ForeignException(getMemberName(), ise));
} catch (KeeperException e) {
 member.controllerConnectionFailure("Failed to get data for new procedure:" + opName,

代码示例来源：origin: apache/hbase

@Override
public void nodeCreated(String path) {
 if (!isInProcedurePath(path)) {
  return;
 }
 LOG.info("Received created event:" + path);
 // if it is a simple start/end/abort then we just rewatch the node
 if (isAcquiredNode(path)) {
  waitForNewProcedures();
  return;
 } else if (isAbortNode(path)) {
  watchForAbortedProcedures();
  return;
 }
 String parent = ZKUtil.getParent(path);
 // if its the end barrier, the procedure can be completed
 if (isReachedNode(parent)) {
  receivedReachedGlobalBarrier(path);
  return;
 } else if (isAbortNode(parent)) {
  abort(path);
  return;
 } else if (isAcquiredNode(parent)) {
  startNewSubprocedure(path);
 } else {
  LOG.debug("Ignoring created notification for node:" + path);
 }
}