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

private void updateZooKeeper(TableState tableState) throws IOException {
  if (tableState == null) {
    return;
  }
  String znode = ZNodePaths.joinZNode(this.master.getZooKeeper().getZNodePaths().tableZNode,
    tableState.getTableName().getNameAsString());
  try {
    // Make sure znode exists.
    if (ZKUtil.checkExists(this.master.getZooKeeper(), znode) == -1) {
      ZKUtil.createAndFailSilent(this.master.getZooKeeper(), znode);
    }
    // Now set newState
    ZooKeeperProtos.DeprecatedTableState.Builder builder =
      ZooKeeperProtos.DeprecatedTableState.newBuilder();
    builder.setState(
      ZooKeeperProtos.DeprecatedTableState.State.valueOf(tableState.getState().toString()));
    byte[] data = ProtobufUtil.prependPBMagic(builder.build().toByteArray());
    ZKUtil.setData(this.master.getZooKeeper(), znode, data);
  } catch (KeeperException e) {
    // Only hbase1 clients suffer if this fails.
    LOG.warn("Failed setting table state to zookeeper mirrored for hbase-1.x clients", e);
  }
}
 

public SplitOrMergeTracker(ZKWatcher watcher, Configuration conf,
                           Abortable abortable) {
  try {
    if (ZKUtil.checkExists(watcher, watcher.getZNodePaths().switchZNode) < 0) {
      ZKUtil.createAndFailSilent(watcher, watcher.getZNodePaths().switchZNode);
    }
  } catch (KeeperException e) {
    throw new RuntimeException(e);
  }
  splitZnode = ZNodePaths.joinZNode(watcher.getZNodePaths().switchZNode,
    conf.get("zookeeper.znode.switch.split", "split"));
  mergeZnode = ZNodePaths.joinZNode(watcher.getZNodePaths().switchZNode,
    conf.get("zookeeper.znode.switch.merge", "merge"));
  splitStateTracker = new SwitchStateTracker(watcher, splitZnode, abortable);
  mergeStateTracker = new SwitchStateTracker(watcher, mergeZnode, abortable);
}
 

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

/**
 * This acts as the ack for a completed procedure
 */
@Override
public void sendMemberCompleted(Subprocedure sub, byte[] data) throws IOException {
  String procName = sub.getName();
  LOG.debug("Marking procedure  '" + procName + "' completed for member '" + memberName
      + "' in zk");
  String joinPath =
    ZNodePaths.joinZNode(zkController.getReachedBarrierNode(procName), memberName);
  // ProtobufUtil.prependPBMagic does not take care of null
  if (data == null) {
    data = new byte[0];
  }
  try {
    ZKUtil.createAndFailSilent(zkController.getWatcher(), joinPath,
      ProtobufUtil.prependPBMagic(data));
  } catch (KeeperException e) {
    member.controllerConnectionFailure("Failed to post zk node:" + joinPath
        + " to join procedure barrier.", e, procName);
  }
}
 

/**
 * This should be called by the member and should write a serialized root cause exception as
 * to the abort znode.
 */
@Override
public void sendMemberAborted(Subprocedure sub, ForeignException ee) {
  if (sub == null) {
    LOG.error("Failed due to null subprocedure", ee);
    return;
  }
  String procName = sub.getName();
  LOG.debug("Aborting procedure (" + procName + ") in zk");
  String procAbortZNode = zkController.getAbortZNode(procName);
  try {
    String source = (ee.getSource() == null) ? memberName: ee.getSource();
    byte[] errorInfo = ProtobufUtil.prependPBMagic(ForeignException.serialize(source, ee));
    ZKUtil.createAndFailSilent(zkController.getWatcher(), procAbortZNode, errorInfo);
    LOG.debug("Finished creating abort znode:" + procAbortZNode);
  } catch (KeeperException e) {
    // possible that we get this error for the procedure if we already reset the zk state, but in
    // that case we should still get an error for that procedure anyways
    zkController.logZKTree(zkController.getBaseZnode());
    member.controllerConnectionFailure("Failed to post zk node:" + procAbortZNode
        + " to abort procedure", e, procName);
  }
}
 

/**
 * This is the abort message being sent by the coordinator to member
 *
 * TODO this code isn't actually used but can be used to issue a cancellation from the
 * coordinator.
 */
@Override
final public void sendAbortToMembers(Procedure proc, ForeignException ee) {
  String procName = proc.getName();
  LOG.debug("Aborting procedure '" + procName + "' in zk");
  String procAbortNode = zkProc.getAbortZNode(procName);
  try {
    LOG.debug("Creating abort znode:" + procAbortNode);
    String source = (ee.getSource() == null) ? coordName : ee.getSource();
    byte[] errorInfo = ProtobufUtil.prependPBMagic(ForeignException.serialize(source, ee));
    // first create the znode for the procedure
    ZKUtil.createAndFailSilent(zkProc.getWatcher(), procAbortNode, errorInfo);
    LOG.debug("Finished creating abort node:" + procAbortNode);
  } catch (KeeperException e) {
    // possible that we get this error for the procedure if we already reset the zk state, but in
    // that case we should still get an error for that procedure anyways
    zkProc.logZKTree(zkProc.baseZNode);
    coordinator.rpcConnectionFailure("Failed to post zk node:" + procAbortNode
        + " to abort procedure '" + procName + "'", new IOException(e));
  }
}
 

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

public void clearZNodes(String procedureName) throws KeeperException {
  LOG.info("Clearing all znodes for procedure " + procedureName + "including nodes "
      + acquiredZnode + " " + reachedZnode + " " + abortZnode);

  // Make sure we trigger the watches on these nodes by creating them. (HBASE-13885)
  String acquiredBarrierNode = getAcquiredBarrierNode(procedureName);
  String reachedBarrierNode = getReachedBarrierNode(procedureName);
  String abortZNode = getAbortZNode(procedureName);

  ZKUtil.createAndFailSilent(watcher, acquiredBarrierNode);
  ZKUtil.createAndFailSilent(watcher, abortZNode);

  ZKUtil.deleteNodeRecursivelyMultiOrSequential(watcher, true, acquiredBarrierNode,
    reachedBarrierNode, abortZNode);

  if (LOG.isTraceEnabled()) {
    logZKTree(this.baseZNode);
  }
}
 

@Before
public void setup() throws Exception {
  TEST_UTIL.startMiniZKCluster();
  Configuration conf = TEST_UTIL.getConfiguration();
  zkw = new ZKWatcher(TEST_UTIL.getConfiguration(),
      "split-log-worker-tests", null);
  ds = new DummyServer(zkw, conf);
  ZKUtil.deleteChildrenRecursively(zkw, zkw.getZNodePaths().baseZNode);
  ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().baseZNode);
  assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().baseZNode), not(is(-1)));
  LOG.debug(zkw.getZNodePaths().baseZNode + " created");
  ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().splitLogZNode);
  assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().splitLogZNode), not(is(-1)));

  LOG.debug(zkw.getZNodePaths().splitLogZNode + " created");
  ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().rsZNode);
  assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().rsZNode), not(is(-1)));

  SplitLogCounters.resetCounters();
  executorService = new ExecutorService("TestSplitLogWorker");
  executorService.startExecutorService(ExecutorType.RS_LOG_REPLAY_OPS, 10);
}
 

@Before
public void setup() throws Exception {
  TEST_UTIL = new HBaseTestingUtility();
  TEST_UTIL.startMiniZKCluster();
  conf = TEST_UTIL.getConfiguration();
  // Use a different ZK wrapper instance for each tests.
  zkw =
      new ZKWatcher(conf, "split-log-manager-tests" + TEST_UTIL.getRandomUUID().toString(), null);
  master = new DummyMasterServices(zkw, conf);

  ZKUtil.deleteChildrenRecursively(zkw, zkw.getZNodePaths().baseZNode);
  ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().baseZNode);
  assertTrue(ZKUtil.checkExists(zkw, zkw.getZNodePaths().baseZNode) != -1);
  LOG.debug(zkw.getZNodePaths().baseZNode + " created");
  ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().splitLogZNode);
  assertTrue(ZKUtil.checkExists(zkw, zkw.getZNodePaths().splitLogZNode) != -1);
  LOG.debug(zkw.getZNodePaths().splitLogZNode + " created");

  resetCounters();

  // By default, we let the test manage the error as before, so the server
  // does not appear as dead from the master point of view, only from the split log pov.
  Mockito.when(sm.isServerOnline(Mockito.any())).thenReturn(true);

  to = 12000;
  conf.setInt(HConstants.HBASE_SPLITLOG_MANAGER_TIMEOUT, to);
  conf.setInt("hbase.splitlog.manager.unassigned.timeout", 2 * to);

  conf.setInt("hbase.splitlog.manager.timeoutmonitor.period", 100);
  to = to + 16 * 100;
}
 

/**
 * create an address tracker instance
 * @param sn if not-null set the active master
 * @param infoPort if there is an active master, set its info port.
 */
private MasterAddressTracker setupMasterTracker(final ServerName sn, final int infoPort)
    throws Exception {
  ZKWatcher zk = new ZKWatcher(TEST_UTIL.getConfiguration(),
      name.getMethodName(), null);
  ZKUtil.createAndFailSilent(zk, zk.getZNodePaths().baseZNode);

  // Should not have a master yet
  MasterAddressTracker addressTracker = new MasterAddressTracker(zk, null);
  addressTracker.start();
  assertFalse(addressTracker.hasMaster());
  zk.registerListener(addressTracker);

  // Use a listener to capture when the node is actually created
  NodeCreationListener listener = new NodeCreationListener(zk,
          zk.getZNodePaths().masterAddressZNode);
  zk.registerListener(listener);

  if (sn != null) {
    LOG.info("Creating master node");
    MasterAddressTracker.setMasterAddress(zk, zk.getZNodePaths().masterAddressZNode,
            sn, infoPort);

    // Wait for the node to be created
    LOG.info("Waiting for master address manager to be notified");
    listener.waitForCreation();
    LOG.info("Master node created");
  }
  return addressTracker;
}
 

/**
 * Perform a best effort enable of hfile retention, which relies on zookeeper communicating the
 * change back to the hfile cleaner.
 * <p>
 * No attempt is made to make sure that backups are successfully created - it is inherently an
 * <b>asynchronous operation</b>.
 * @param zooKeeper watcher connection to zk cluster
 * @param table table name on which to enable archiving
 * @throws KeeperException if a ZooKeeper operation fails
 */
private void enable(ZKWatcher zooKeeper, byte[] table) throws KeeperException {
  LOG.debug("Ensuring archiving znode exists");
  ZKUtil.createAndFailSilent(zooKeeper, archiveZnode);

  // then add the table to the list of znodes to archive
  String tableNode = this.getTableNode(table);
  LOG.debug("Creating: " + tableNode + ", data: []");
  ZKUtil.createSetData(zooKeeper, tableNode, new byte[0]);
}