Java 类org.apache.hadoop.hbase.util.EnvironmentEdgeManager 实例源码

/**
 * Check if the call did timeout. Set an exception (includes a notify) if it's the case.
 * @return true if the call is on timeout, false otherwise.
 */
public boolean checkAndSetTimeout() {
  if (timeout == 0){
    return false;
  }

  long waitTime = EnvironmentEdgeManager.currentTime() - getStartTime();
  if (waitTime >= timeout) {
    IOException ie = new CallTimeoutException("Call id=" + id +
        ", waitTime=" + waitTime + ", operationTimeout=" + timeout + " expired.");
    setException(ie); // includes a notify
    return true;
  } else {
    return false;
  }
}

synchronized void removeExpiredKeys() {
  if (!leaderElector.isMaster()) {
    LOG.info("Skipping removeExpiredKeys() because not running as master.");
    return;
  }

  long now = EnvironmentEdgeManager.currentTime();
  Iterator<AuthenticationKey> iter = allKeys.values().iterator();
  while (iter.hasNext()) {
    AuthenticationKey key = iter.next();
    if (key.getExpiration() < now) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Removing expired key "+key.getKeyId());
      }
      iter.remove();
      zkWatcher.removeKeyFromZK(key);
    }
  }
}

synchronized void rollCurrentKey() {
  if (!leaderElector.isMaster()) {
    LOG.info("Skipping rollCurrentKey() because not running as master.");
    return;
  }

  long now = EnvironmentEdgeManager.currentTime();
  AuthenticationKey prev = currentKey;
  AuthenticationKey newKey = new AuthenticationKey(++idSeq,
      Long.MAX_VALUE, // don't allow to expire until it's replaced by a new key
      generateSecret());
  allKeys.put(newKey.getKeyId(), newKey);
  currentKey = newKey;
  zkWatcher.addKeyToZK(newKey);
  lastKeyUpdate = now;

  if (prev != null) {
    // make sure previous key is still stored
    prev.setExpiration(now + tokenMaxLifetime);
    allKeys.put(prev.getKeyId(), prev);
    zkWatcher.updateKeyInZK(prev);
  }
}

private void heartbeat(String path, int new_version, ServerName workerName) {
  Task task = findOrCreateOrphanTask(path);
  if (new_version != task.last_version) {
    if (task.isUnassigned()) {
      LOG.info("task " + path + " acquired by " + workerName);
    }
    task.heartbeat(EnvironmentEdgeManager.currentTime(), new_version, workerName);
    SplitLogCounters.tot_mgr_heartbeat.incrementAndGet();
  } else {
    // duplicate heartbeats - heartbeats w/o zk node version
    // changing - are possible. The timeout thread does
    // getDataSetWatch() just to check whether a node still
    // exists or not
  }
  return;
}

@Override
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "GC_UNRELATED_TYPES",
    justification = "I do not understand why the complaints, it looks good to me -- FIX")
protected void chore() {
  // Prefetch online tables/namespaces
  for (TableName table : QuotaCache.this.rsServices.getOnlineTables()) {
    if (table.isSystemTable()) continue;
    if (!QuotaCache.this.tableQuotaCache.contains(table)) {
      QuotaCache.this.tableQuotaCache.putIfAbsent(table, new QuotaState());
    }
    String ns = table.getNamespaceAsString();
    if (!QuotaCache.this.namespaceQuotaCache.contains(ns)) {
      QuotaCache.this.namespaceQuotaCache.putIfAbsent(ns, new QuotaState());
    }
  }

  fetchNamespaceQuotaState();
  fetchTableQuotaState();
  fetchUserQuotaState();
  lastUpdate = EnvironmentEdgeManager.currentTime();
}

@Override
public long refill(long limit) {
  final long now = EnvironmentEdgeManager.currentTime();
  if (nextRefillTime == -1) {
    // Till now no resource has been consumed.
    nextRefillTime = EnvironmentEdgeManager.currentTime();
    return limit;
  }

  long delta = (limit * (now - nextRefillTime)) / super.getTimeUnitInMillis();
  if (delta > 0) {
    this.nextRefillTime = now;
    return Math.min(limit, delta);
  }
  return 0;
}

private InterProcessLock createTableLock() {
  String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode,
      tableName.getNameAsString());

  ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder()
    .setTableName(ProtobufUtil.toProtoTableName(tableName))
    .setLockOwner(ProtobufUtil.toServerName(serverName))
    .setThreadId(Thread.currentThread().getId())
    .setPurpose(purpose)
    .setIsShared(isShared)
    .setCreateTime(EnvironmentEdgeManager.currentTime()).build();
  byte[] lockMetadata = toBytes(data);

  InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode,
    METADATA_HANDLER);
  return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata);
}

/**
 * This method is the base split method that splits WAL files matching a filter. Callers should
 * pass the appropriate filter for meta and non-meta WALs.
 * @param serverNames logs belonging to these servers will be split; this will rename the log
 *                    directory out from under a soft-failed server
 * @param filter
 * @throws IOException
 */
public void splitLog(final Set<ServerName> serverNames, PathFilter filter) throws IOException {
  long splitTime = 0, splitLogSize = 0;
  List<Path> logDirs = getLogDirs(serverNames);

  splitLogManager.handleDeadWorkers(serverNames);
  splitTime = EnvironmentEdgeManager.currentTime();
  splitLogSize = splitLogManager.splitLogDistributed(serverNames, logDirs, filter);
  splitTime = EnvironmentEdgeManager.currentTime() - splitTime;

  if (this.metricsMasterFilesystem != null) {
    if (filter == META_FILTER) {
      this.metricsMasterFilesystem.addMetaWALSplit(splitTime, splitLogSize);
    } else {
      this.metricsMasterFilesystem.addSplit(splitTime, splitLogSize);
    }
  }
}

@Override
protected boolean waitUntilDone(long timeout) throws InterruptedException {
  long startTime = EnvironmentEdgeManager.currentTime();
  long remaining = timeout;
  List<HRegionInfo> regions = null;
  long lastLogTime = startTime;
  while (!server.isStopped() && remaining > 0) {
    Thread.sleep(waitingTimeForEvents);
    regions = assignmentManager.getRegionStates().getRegionsOfTable(tableName);
    long now = EnvironmentEdgeManager.currentTime();
    // Don't log more than once every ten seconds. Its obnoxious. And only log table regions
    // if we are waiting a while for them to go down...
    if (LOG.isDebugEnabled() && ((now - lastLogTime) > 10000)) {
      lastLogTime = now;
      LOG.debug("Disable waiting until done; " + remaining + " ms remaining; " + regions);
    }
    if (regions.isEmpty()) break;
    remaining = timeout - (now - startTime);
  }
  return regions != null && regions.isEmpty();
}

public static <T> T waitFor(MasterProcedureEnv env, long waitTime, long waitingTimeForEvents,
    String purpose, Predicate<T> predicate) throws IOException {
  final long done = EnvironmentEdgeManager.currentTime() + waitTime;
  do {
    T result = predicate.evaluate();
    if (result != null && !result.equals(Boolean.FALSE)) {
      return result;
    }
    try {
      Thread.sleep(waitingTimeForEvents);
    } catch (InterruptedException e) {
      LOG.warn("Interrupted while sleeping, waiting on " + purpose);
      throw (InterruptedIOException)new InterruptedIOException().initCause(e);
    }
    LOG.debug("Waiting on " + purpose);
  } while (EnvironmentEdgeManager.currentTime() < done && env.isRunning());

  throw new TimeoutIOException("Timed out while waiting on " + purpose);
}

/**
 * Wait on region to clear regions-in-transition or time out
 * @param hri
 * @param timeOut Milliseconds to wait for current region to be out of transition state.
 * @return True when a region clears regions-in-transition before timeout otherwise false
 * @throws InterruptedException
 */
public boolean waitOnRegionToClearRegionsInTransition(final HRegionInfo hri, long timeOut)
    throws InterruptedException {
  if (!regionStates.isRegionInTransition(hri)) return true;
  long end = (timeOut <= 0) ? Long.MAX_VALUE : EnvironmentEdgeManager.currentTime()
      + timeOut;
  // There is already a timeout monitor on regions in transition so I
  // should not have to have one here too?
  LOG.info("Waiting for " + hri.getEncodedName() +
      " to leave regions-in-transition, timeOut=" + timeOut + " ms.");
  while (!this.server.isStopped() && regionStates.isRegionInTransition(hri)) {
    regionStates.waitForUpdate(100);
    if (EnvironmentEdgeManager.currentTime() > end) {
      LOG.info("Timed out on waiting for " + hri.getEncodedName() + " to be assigned.");
      return false;
    }
  }
  if (this.server.isStopped()) {
    LOG.info("Giving up wait on regions in transition because stoppable.isStopped is set");
    return false;
  }
  return true;
}

/**
 * Archive the store file
 * @param fs the filesystem where the store files live
 * @param regionInfo region hosting the store files
 * @param conf {@link Configuration} to examine to determine the archive directory
 * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
 * @param family the family hosting the store files
 * @param storeFile file to be archived
 * @throws IOException if the files could not be correctly disposed.
 */
public static void archiveStoreFile(Configuration conf, FileSystem fs, HRegionInfo regionInfo,
    Path tableDir, byte[] family, Path storeFile) throws IOException {
  Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family);
  // make sure we don't archive if we can't and that the archive dir exists
  if (!fs.mkdirs(storeArchiveDir)) {
    throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:"
        + Bytes.toString(family) + ", deleting compacted files instead.");
  }

  // do the actual archive
  long start = EnvironmentEdgeManager.currentTime();
  File file = new FileablePath(fs, storeFile);
  if (!resolveAndArchiveFile(storeArchiveDir, file, Long.toString(start))) {
    throw new IOException("Failed to archive/delete the file for region:"
        + regionInfo.getRegionNameAsString() + ", family:" + Bytes.toString(family)
        + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.");
  }
}

@Test
public void testCanExecuteOfAverageIntervalRateLimiter() throws InterruptedException {
  RateLimiter limiter = new AverageIntervalRateLimiter();
  // when set limit is 100 per sec, this AverageIntervalRateLimiter will support at max 200 per sec
  limiter.set(100, TimeUnit.SECONDS);
  limiter.setNextRefillTime(EnvironmentEdgeManager.currentTime());
  assertEquals(50, testCanExecuteByRate(limiter, 50));

  // refill the avail to limit
  limiter.set(100, TimeUnit.SECONDS);
  limiter.setNextRefillTime(EnvironmentEdgeManager.currentTime());
  assertEquals(100, testCanExecuteByRate(limiter, 100));

  // refill the avail to limit
  limiter.set(100, TimeUnit.SECONDS);
  limiter.setNextRefillTime(EnvironmentEdgeManager.currentTime());
  assertEquals(200, testCanExecuteByRate(limiter, 200));

  // refill the avail to limit
  limiter.set(100, TimeUnit.SECONDS);
  limiter.setNextRefillTime(EnvironmentEdgeManager.currentTime());
  assertEquals(200, testCanExecuteByRate(limiter, 500));
}

/**
 * Create a generic timer for a task/process.
 * @param listener listener to notify if the process times out
 * @param maxTime max allowed running time for the process. Timer starts on calls to
 *          {@link #start()}
 */
public TimeoutExceptionInjector(final ForeignExceptionListener listener, final long maxTime) {
  this.maxTime = maxTime;
  timer = new Timer();
  timerTask = new TimerTask() {
    @Override
    public void run() {
      // ensure we don't run this task multiple times
      synchronized (this) {
        // quick exit if we already marked the task complete
        if (TimeoutExceptionInjector.this.complete) return;
        // mark the task is run, to avoid repeats
        TimeoutExceptionInjector.this.complete = true;
      }
      long end = EnvironmentEdgeManager.currentTime();
      TimeoutException tee =  new TimeoutException(
          "Timeout caused Foreign Exception", start, end, maxTime);
      String source = "timer-" + timer;
      listener.receive(new ForeignException(source, tee));
    }
  };
}

@Test
public void testSortExtract(){
  ManualEnvironmentEdge mee = new ManualEnvironmentEdge();
  EnvironmentEdgeManager.injectEdge(mee);
  mee.setValue(1);

  DeadServer d = new DeadServer();


  d.add(hostname123);
  mee.incValue(1);
  d.add(hostname1234);
  mee.incValue(1);
  d.add(hostname12345);

  List<Pair<ServerName, Long>> copy = d.copyDeadServersSince(2L);
  Assert.assertEquals(2, copy.size());

  Assert.assertEquals(hostname1234, copy.get(0).getFirst());
  Assert.assertEquals(new Long(2L), copy.get(0).getSecond());

  Assert.assertEquals(hostname12345, copy.get(1).getFirst());
  Assert.assertEquals(new Long(3L), copy.get(1).getSecond());

  EnvironmentEdgeManager.reset();
}

private  boolean hasExpiredStores(Collection<StoreFile> files) {
  long currentTime = EnvironmentEdgeManager.currentTime();
  for(StoreFile sf: files){
    // Check MIN_VERSIONS is in HStore removeUnneededFiles
    Long maxTs = sf.getReader().getMaxTimestamp();
    long maxTtl = storeConfigInfo.getStoreFileTtl();
    if(maxTs == null 
        || maxTtl == Long.MAX_VALUE
        || (currentTime - maxTtl < maxTs)){
      continue; 
    } else{
      return true;
    }
  }
  return false;
}

private  Collection<StoreFile> getExpiredStores(Collection<StoreFile> files,
  Collection<StoreFile> filesCompacting) {
  long currentTime = EnvironmentEdgeManager.currentTime();
  Collection<StoreFile> expiredStores = new ArrayList<StoreFile>();    
  for(StoreFile sf: files){
    // Check MIN_VERSIONS is in HStore removeUnneededFiles
    Long maxTs = sf.getReader().getMaxTimestamp();
    long maxTtl = storeConfigInfo.getStoreFileTtl();
    if(maxTs == null 
        || maxTtl == Long.MAX_VALUE
        || (currentTime - maxTtl < maxTs)){
      continue; 
    } else if(filesCompacting == null || filesCompacting.contains(sf) == false){
      expiredStores.add(sf);
    }
  }
  return expiredStores;
}

private void checkShouldFlush(Configuration conf, boolean expected) throws Exception {
  try {
    EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
    EnvironmentEdgeManager.injectEdge(edge);
    HBaseTestingUtility hbaseUtility = HBaseTestingUtility.createLocalHTU(conf);
    HRegion region = hbaseUtility.createTestRegion("foobar", new HColumnDescriptor("foo"));

    List<Store> stores = region.getStores();
    assertTrue(stores.size() == 1);

    Store s = stores.iterator().next();
    edge.setCurrentTimeMillis(1234);
    s.add(KeyValueTestUtil.create("r", "f", "q", 100, "v"));
    edge.setCurrentTimeMillis(1234 + 100);
    StringBuffer sb = new StringBuffer();
    assertTrue(region.shouldFlush(sb) == false);
    edge.setCurrentTimeMillis(1234 + 10000);
    assertTrue(region.shouldFlush(sb) == expected);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}

/**
 * Reports the operation from WAL during replay.
 * @param group Nonce group.
 * @param nonce Nonce.
 * @param writeTime Entry write time, used to ignore entries that are too old.
 */
public void reportOperationFromWal(long group, long nonce, long writeTime) {
  if (nonce == HConstants.NO_NONCE) return;
  // Give the write time some slack in case the clocks are not synchronized.
  long now = EnvironmentEdgeManager.currentTime();
  if (now > writeTime + (deleteNonceGracePeriod * 1.5)) return;
  OperationContext newResult = new OperationContext();
  newResult.setState(OperationContext.DONT_PROCEED);
  NonceKey nk = new NonceKey(group, nonce);
  OperationContext oldResult = nonces.putIfAbsent(nk, newResult);
  if (oldResult != null) {
    // Some schemes can have collisions (for example, expiring hashes), so just log it.
    // We have no idea about the semantics here, so this is the least of many evils.
    LOG.warn("Nonce collision during WAL recovery: " + nk
        + ", " + oldResult + " with " + newResult);
  }
}

/**
 * Sleep for an expected amount of time.
 * <p>
 * This is nearly a copy of what the Sleeper does, but with the ability to know if you
 * got interrupted while sleeping.
 * </p>
 *
 * @return <tt>true</tt> if the sleep completely entirely successfully,
 * but otherwise <tt>false</tt> if the sleep was interrupted.
 */
private boolean sleep() {
  long now = EnvironmentEdgeManager.currentTime();
  long startTime = now;
  long waitTime = sleepTime;
  while (waitTime > 0) {
    long woke = -1;
    try {
      synchronized (sleepLock) {
        if (triggerWake) break;
        sleepLock.wait(waitTime);
      }
      woke = EnvironmentEdgeManager.currentTime();
    } catch (InterruptedException iex) {
      return false;
    }
    // Recalculate waitTime.
    woke = (woke == -1) ? EnvironmentEdgeManager.currentTime() : woke;
    waitTime = waitTime - (woke - startTime);
  }
  return true;
}

/**
 * Updates the location of the specified region to be the specified server.
 * <p>
 * Connects to the specified server which should be hosting the specified
 * catalog region name to perform the edit.
 *
 * @param connection connection we're using
 * @param regionInfo region to update location of
 * @param sn Server name
 * @param openSeqNum the latest sequence number obtained when the region was open
 * @param masterSystemTime wall clock time from master if passed in the open region RPC or -1
 * @throws IOException In particular could throw {@link java.net.ConnectException}
 * if the server is down on other end.
 */
private static void updateLocation(final Connection connection,
                                   HRegionInfo regionInfo, ServerName sn, long openSeqNum,
                                   long masterSystemTime)
  throws IOException {

  // use the maximum of what master passed us vs local time.
  long time = Math.max(EnvironmentEdgeManager.currentTime(), masterSystemTime);

  // region replicas are kept in the primary region's row
  Put put = new Put(getMetaKeyForRegion(regionInfo), time);
  addLocation(put, sn, openSeqNum, time, regionInfo.getReplicaId());
  putToMetaTable(connection, put);
  LOG.info("Updated row " + regionInfo.getRegionNameAsString() +
    " with server=" + sn);
}

/**
 * Handles failures encountered when communicating with a server.
 *
 * Updates the FailureInfo in repeatedFailuresMap to reflect the failure.
 * Throws RepeatedConnectException if the client is in Fast fail mode.
 *
 * @param serverName
 * @param t
 *          - the throwable to be handled.
 * @throws PreemptiveFastFailException
 */
private void handleFailureToServer(ServerName serverName, Throwable t) {
  if (serverName == null || t == null) {
    return;
  }
  long currentTime = EnvironmentEdgeManager.currentTime();
  FailureInfo fInfo = repeatedFailuresMap.get(serverName);
  if (fInfo == null) {
    fInfo = new FailureInfo(currentTime);
    FailureInfo oldfInfo = repeatedFailuresMap.putIfAbsent(serverName, fInfo);

    if (oldfInfo != null) {
      fInfo = oldfInfo;
    }
  }
  fInfo.timeOfLatestAttemptMilliSec = currentTime;
  fInfo.numConsecutiveFailures.incrementAndGet();
}

@Test
public void testMaxSend() {
  ClusterStatusPublisher csp = new ClusterStatusPublisher() {
    @Override
    protected List<Pair<ServerName, Long>> getDeadServers(long since) {
      List<Pair<ServerName, Long>> res = new ArrayList<Pair<ServerName, Long>>();
      switch ((int) EnvironmentEdgeManager.currentTime()) {
        case 2:
          res.add(new Pair<ServerName, Long>(ServerName.valueOf("hn", 10, 10), 1L));
          break;
        case 1000:
          break;
      }

      return res;
    }
  };

  mee.setValue(2);
  for (int i = 0; i < ClusterStatusPublisher.NB_SEND; i++) {
    Assert.assertEquals("i=" + i, 1, csp.generateDeadServersListToSend().size());
  }
  mee.setValue(1000);
  Assert.assertTrue(csp.generateDeadServersListToSend().isEmpty());
}

public void testShouldFlushMeta() throws Exception {
  // write an edit in the META and ensure the shouldFlush (that the periodic memstore
  // flusher invokes) returns true after SYSTEM_CACHE_FLUSH_INTERVAL (even though
  // the MEMSTORE_PERIODIC_FLUSH_INTERVAL is set to a higher value)
  Configuration conf = new Configuration();
  conf.setInt(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, HRegion.SYSTEM_CACHE_FLUSH_INTERVAL * 10);
  HBaseTestingUtility hbaseUtility = HBaseTestingUtility.createLocalHTU(conf);
  Path testDir = hbaseUtility.getDataTestDir();
  EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
  EnvironmentEdgeManager.injectEdge(edge);
  edge.setCurrentTimeMillis(1234);
  WALFactory wFactory = new WALFactory(conf, null, "1234");
  HRegion meta = HRegion.createHRegion(HRegionInfo.FIRST_META_REGIONINFO, testDir,
      conf, HTableDescriptor.metaTableDescriptor(conf),
      wFactory.getMetaWAL(HRegionInfo.FIRST_META_REGIONINFO.
          getEncodedNameAsBytes()));
  HRegionInfo hri = new HRegionInfo(TableName.valueOf("testShouldFlushMeta"),
      Bytes.toBytes("row_0200"), Bytes.toBytes("row_0300"));
  HTableDescriptor desc = new HTableDescriptor(TableName.valueOf("testShouldFlushMeta"));
  desc.addFamily(new HColumnDescriptor("foo".getBytes()));
  HRegion r =
      HRegion.createHRegion(hri, testDir, conf, desc,
          wFactory.getWAL(hri.getEncodedNameAsBytes()));
  HRegion.addRegionToMETA(meta, r);
  edge.setCurrentTimeMillis(1234 + 100);
  StringBuffer sb = new StringBuffer();
  assertTrue(meta.shouldFlush(sb) == false);
  edge.setCurrentTimeMillis(edge.currentTime() + HRegion.SYSTEM_CACHE_FLUSH_INTERVAL + 1);
  assertTrue(meta.shouldFlush(sb) == true);
}

/**
 * Get how long the caller should sleep according to the current size and
 * current cycle's total push size and start tick, return the sleep interval
 * for throttling control.
 * @param size is the size of edits to be pushed
 * @return sleep interval for throttling control
 */
public long getNextSleepInterval(final int size) {
  if (!this.enabled) {
    return 0;
  }

  long sleepTicks = 0;
  long now = EnvironmentEdgeManager.currentTime();
  // 1. if cyclePushSize exceeds bandwidth, we need to sleep some
  //    following cycles to amortize, this case can occur when a single push
  //    exceeds the bandwidth
  if ((double)this.cyclePushSize > bandwidth) {
    double cycles = Math.ceil((double)this.cyclePushSize / bandwidth);
    long shouldTillTo = this.cycleStartTick + (long)(cycles * 100);
    if (shouldTillTo > now) {
      sleepTicks = shouldTillTo - now;
    } else {
      // no reset in shipEdits since no sleep, so we need to reset cycleStartTick here!
      this.cycleStartTick = now;
    }
    this.cyclePushSize = 0;
  } else {
    long nextCycleTick = this.cycleStartTick + 100;  //a cycle is 100ms
    if (now >= nextCycleTick) {
      // 2. switch to next cycle if the current cycle has passed
      this.cycleStartTick = now;
      this.cyclePushSize = 0;
    } else if (this.cyclePushSize > 0 &&
        (double)(this.cyclePushSize + size) >= bandwidth) {
      // 3. delay the push to next cycle if exceeds throttling bandwidth.
      //    enforcing cyclePushSize > 0 to avoid the unnecessary sleep for case
      //    where a cycle's first push size(currentSize) > bandwidth
      sleepTicks = nextCycleTick - now;
      this.cyclePushSize = 0;
    }
  }
  return sleepTicks;
}

@Override
public String toString() {
  StringBuilder sb = new StringBuilder();
  sb.append("Procedure=");
  sb.append(procName);
  sb.append(" (id=");
  sb.append(procId);
  if (hasParentId()) {
    sb.append(", parent=");
    sb.append(parentId);
  }
  if (hasOwner()) {
    sb.append(", owner=");
    sb.append(procOwner);
  }
  sb.append(", state=");
  sb.append(procState);

  long now = EnvironmentEdgeManager.currentTime();
  sb.append(", startTime=");
  sb.append(StringUtils.formatTime(now - startTime));
  sb.append(" ago, lastUpdate=");
  sb.append(StringUtils.formatTime(now - startTime));
  sb.append(" ago");

  if (isFailed()) {
    sb.append(", exception=\"");
    sb.append(getExceptionMessage());
    sb.append("\"");
  }
  sb.append(")");
  return sb.toString();
}

/**
 * Set the age of the last edit that was shipped
 * @param timestamp write time of the edit
 * @param walGroup which group we are setting
 */
public void setAgeOfLastShippedOp(long timestamp, String walGroup) {
  long age = EnvironmentEdgeManager.currentTime() - timestamp;
  singleSourceSource.setLastShippedAge(age);
  globalSourceSource.setLastShippedAge(age);
  this.lastTimeStamps.put(walGroup, timestamp);
}

@Override
protected synchronized byte[] createPassword(AuthenticationTokenIdentifier identifier) {
  long now = EnvironmentEdgeManager.currentTime();
  AuthenticationKey secretKey = currentKey;
  identifier.setKeyId(secretKey.getKeyId());
  identifier.setIssueDate(now);
  identifier.setExpirationDate(now + tokenMaxLifetime);
  identifier.setSequenceNumber(tokenSeq.getAndIncrement());
  return createPassword(identifier.getBytes(),
      secretKey.getKey());
}

@Override
public byte[] retrievePassword(AuthenticationTokenIdentifier identifier)
    throws InvalidToken {
  long now = EnvironmentEdgeManager.currentTime();
  if (identifier.getExpirationDate() < now) {
    throw new InvalidToken("Token has expired");
  }
  AuthenticationKey masterKey = allKeys.get(identifier.getKeyId());
  if(masterKey == null) {
    if(zkWatcher.getWatcher().isAborted()) {
      LOG.error("ZookeeperWatcher is abort");
      throw new InvalidToken("Token keys could not be sync from zookeeper"
          + " because of ZookeeperWatcher abort");
    }
    synchronized (this) {
      if (!leaderElector.isAlive() || leaderElector.isStopped()) {
        LOG.warn("Thread leaderElector[" + leaderElector.getName() + ":"
            + leaderElector.getId() + "] is stoped or not alive");
        leaderElector.start();
        LOG.info("Thread leaderElector [" + leaderElector.getName() + ":"
            + leaderElector.getId() + "] is started");
      }
    }
    zkWatcher.refreshKeys();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Sync token keys from zookeeper");
    }
    masterKey = allKeys.get(identifier.getKeyId());
  }
  if (masterKey == null) {
    throw new InvalidToken("Unknown master key for token (id="+
        identifier.getKeyId()+")");
  }
  // regenerate the password
  return createPassword(identifier.getBytes(),
      masterKey.getKey());
}

/**
 * Make a blocking call. Throws exceptions if there are network problems or if the remote code
 * threw an exception.
 *
 * @param ticket Be careful which ticket you pass. A new user will mean a new Connection.
 *               {@link UserProvider#getCurrent()} makes a new instance of User each time so
 *               will be a
 *               new Connection each time.
 * @return A pair with the Message response and the Cell data (if any).
 */
Message callBlockingMethod(Descriptors.MethodDescriptor md, PayloadCarryingRpcController pcrc,
    Message param, Message returnType, final User ticket, final InetSocketAddress isa)
    throws ServiceException {
  if (pcrc == null) {
    pcrc = new PayloadCarryingRpcController();
  }

  Pair<Message, CellScanner> val;
  try {
    final MetricsConnection.CallStats cs = MetricsConnection.newCallStats();
    cs.setStartTime(EnvironmentEdgeManager.currentTime());
    val = call(pcrc, md, param, returnType, ticket, isa, cs);
    // Shove the results into controller so can be carried across the proxy/pb service void.
    pcrc.setCellScanner(val.getSecond());

    cs.setCallTimeMs(EnvironmentEdgeManager.currentTime() - cs.getStartTime());
    if (metrics != null) {
      metrics.updateRpc(md, param, cs);
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Call: " + md.getName() + ", callTime: " + cs.getCallTimeMs() + "ms");
    }
    return val.getFirst();
  } catch (Throwable e) {
    throw new ServiceException(e);
  }
}

@Override
public void nodeDataChanged(String path) {
  Task task;
  task = details.getTasks().get(path);
  if (task != null || ZKSplitLog.isRescanNode(watcher, path)) {
    if (task != null) {
      task.heartbeatNoDetails(EnvironmentEdgeManager.currentTime());
    }
    getDataSetWatch(path, zkretries);
  }
}

/**
 * Wait for the table to be enabled and available
 * If enabling the table exceeds the retry period, an exception is thrown.
 * @param tableName name of the table
 * @throws IOException if a remote or network exception occurs or
 *    table is not enabled after the retries period.
 */
private void waitUntilTableIsEnabled(final TableName tableName) throws IOException {
  boolean enabled = false;
  long start = EnvironmentEdgeManager.currentTime();
  for (int tries = 0; tries < (this.numRetries * this.retryLongerMultiplier); tries++) {
    try {
      enabled = isTableEnabled(tableName);
    } catch (TableNotFoundException tnfe) {
      // wait for table to be created
      enabled = false;
    }
    enabled = enabled && isTableAvailable(tableName);
    if (enabled) {
      break;
    }
    long sleep = getPauseTime(tries);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Sleeping= " + sleep + "ms, waiting for all regions to be " +
        "enabled in " + tableName);
    }
    try {
      Thread.sleep(sleep);
    } catch (InterruptedException e) {
      // Do this conversion rather than let it out because do not want to
      // change the method signature.
      throw (InterruptedIOException)new InterruptedIOException("Interrupted").initCause(e);
    }
  }
  if (!enabled) {
    long msec = EnvironmentEdgeManager.currentTime() - start;
    throw new IOException("Table '" + tableName +
      "' not yet enabled, after " + msec + "ms.");
  }
}

/**
 * Called by the ProcedureExecutor to assign the ID to the newly created procedure.
 */
@VisibleForTesting
@InterfaceAudience.Private
protected void setProcId(final long procId) {
  this.procId = procId;
  this.startTime = EnvironmentEdgeManager.currentTime();
  setState(ProcedureState.RUNNABLE);
}

@Override
public long getWaitInterval(long limit, long available, long amount) {
  if (nextRefillTime == -1) {
    return 0;
  }
  final long now = EnvironmentEdgeManager.currentTime();
  final long refillTime = nextRefillTime;
  return refillTime - now;
}

/**
 * After adding data to the table start a mr job to
 * @throws IOException
 * @throws ClassNotFoundException
 * @throws InterruptedException
 */
private void runCheck() throws IOException, ClassNotFoundException, InterruptedException {
  LOG.info("Running check");
  Configuration conf = getConf();
  String jobName = getTablename() + "_check" + EnvironmentEdgeManager.currentTime();
  Path p = util.getDataTestDirOnTestFS(jobName);

  Job job = new Job(conf);
  job.setJarByClass(getClass());
  job.setJobName(jobName);

  job.setPartitionerClass(NaturalKeyPartitioner.class);
  job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
  job.setSortComparatorClass(CompositeKeyComparator.class);

  Scan scan = new Scan();
  scan.addFamily(CHAIN_FAM);
  scan.addFamily(SORT_FAM);
  scan.setMaxVersions(1);
  scan.setCacheBlocks(false);
  scan.setBatch(1000);

  int replicaCount = conf.getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
  if (replicaCount != NUM_REPLICA_COUNT_DEFAULT) {
    scan.setConsistency(Consistency.TIMELINE);
  }

  TableMapReduceUtil.initTableMapperJob(
      getTablename().getName(),
      scan,
      LinkedListCheckingMapper.class,
      LinkKey.class,
      LinkChain.class,
      job
  );

  job.setReducerClass(LinkedListCheckingReducer.class);
  job.setOutputKeyClass(NullWritable.class);
  job.setOutputValueClass(NullWritable.class);

  FileOutputFormat.setOutputPath(job, p);

  assertEquals(true, job.waitForCompletion(true));

  // Delete the files.
  util.getTestFileSystem().delete(p, true);
}

public void setClusterStatus(ClusterStatus status) {
  long currentTime = EnvironmentEdgeManager.currentTime();
  this.status = status;
  if (currentTime > lastFullRefresh + (CACHE_TIME / 2)) {
    // Only count the refresh if it includes user tables ( eg more than meta and namespace ).
    lastFullRefresh = scheduleFullRefresh()?currentTime:lastFullRefresh;
  }

}

@Override
protected void chore() {
  if (!connected) {
    return;
  }

  List<ServerName> sns = generateDeadServersListToSend();
  if (sns.isEmpty()) {
    // Nothing to send. Done.
    return;
  }

  final long curTime = EnvironmentEdgeManager.currentTime();
  if (lastMessageTime > curTime - messagePeriod) {
    // We already sent something less than 10 second ago. Done.
    return;
  }

  // Ok, we're going to send something then.
  lastMessageTime = curTime;

  // We're reusing an existing protobuf message, but we don't send everything.
  // This could be extended in the future, for example if we want to send stuff like the
  //  hbase:meta server name.
  ClusterStatus cs = new ClusterStatus(VersionInfo.getVersion(),
      master.getMasterFileSystem().getClusterId().toString(),
      null,
      sns,
      master.getServerName(),
      null,
      null,
      null,
      null);


  publisher.publish(cs);
}

/**
 * Create the dead server to send. A dead server is sent NB_SEND times. We send at max
 * MAX_SERVER_PER_MESSAGE at a time. if there are too many dead servers, we send the newly
 * dead first.
 */
protected List<ServerName> generateDeadServersListToSend() {
  // We're getting the message sent since last time, and add them to the list
  long since = EnvironmentEdgeManager.currentTime() - messagePeriod * 2;
  for (Pair<ServerName, Long> dead : getDeadServers(since)) {
    lastSent.putIfAbsent(dead.getFirst(), 0);
  }

  // We're sending the new deads first.
  List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>();
  entries.addAll(lastSent.entrySet());
  Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() {
    @Override
    public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) {
      return o1.getValue().compareTo(o2.getValue());
    }
  });

  // With a limit of MAX_SERVER_PER_MESSAGE
  int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size();
  List<ServerName> res = new ArrayList<ServerName>(max);

  for (int i = 0; i < max; i++) {
    Map.Entry<ServerName, Integer> toSend = entries.get(i);
    if (toSend.getValue() >= (NB_SEND - 1)) {
      lastSent.remove(toSend.getKey());
    } else {
      lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1);
    }

    res.add(toSend.getKey());
  }

  return res;
}

/**
 * Remove the sentinels that are marked as finished and the completion time
 * has exceeded the removal timeout.
 * @param sentinels map of sentinels to clean
 */
private synchronized void cleanupSentinels(final Map<TableName, SnapshotSentinel> sentinels) {
  long currentTime = EnvironmentEdgeManager.currentTime();
  Iterator<Map.Entry<TableName, SnapshotSentinel>> it =
      sentinels.entrySet().iterator();
  while (it.hasNext()) {
    Map.Entry<TableName, SnapshotSentinel> entry = it.next();
    SnapshotSentinel sentinel = entry.getValue();
    if (sentinel.isFinished() &&
        (currentTime - sentinel.getCompletionTimestamp()) > SNAPSHOT_SENTINELS_CLEANUP_TIMEOUT)
    {
      it.remove();
    }
  }
}

public void start() throws IOException {
  if (!MetaTableAccessor.tableExists(masterServices.getConnection(),
      TableName.NAMESPACE_TABLE_NAME)) {
    LOG.info("Namespace table not found. Creating...");
    createNamespaceTable(masterServices);
  }

  try {
    // Wait for the namespace table to be assigned.
    // If timed out, we will move ahead without initializing it.
    // So that it should be initialized later on lazily.
    long startTime = EnvironmentEdgeManager.currentTime();
    int timeout = conf.getInt(NS_INIT_TIMEOUT, DEFAULT_NS_INIT_TIMEOUT);
    while (!isTableAssigned()) {
      if (EnvironmentEdgeManager.currentTime() - startTime + 100 > timeout) {
        // We can't do anything if ns is not online.
        throw new IOException("Timedout " + timeout + "ms waiting for namespace table to " +
          "be assigned");
      }
      Thread.sleep(100);
    }
  } catch (InterruptedException e) {
    throw (InterruptedIOException)new InterruptedIOException().initCause(e);
  }

  // initialize namespace table
  isTableAvailableAndInitialized();
}