import org.apache.htrace.Trace; //导入依赖的package包/类
private Callable<ByteBuffer> getFromOneDataNode(final DNAddrPair datanode,
    final LocatedBlock block, final long start, final long end,
    final ByteBuffer bb,
    final Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap,
    final int hedgedReadId) {
  final Span parentSpan = Trace.currentSpan();
  return new Callable<ByteBuffer>() {
    @Override
    public ByteBuffer call() throws Exception {
      byte[] buf = bb.array();
      int offset = bb.position();
      TraceScope scope =
          Trace.startSpan("hedgedRead" + hedgedReadId, parentSpan);
      try {
        actualGetFromOneDataNode(datanode, block, start, end, buf, offset,
            corruptedBlockMap);
        return bb;
      } finally {
        scope.close();
      }
    }
  };
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@Override
public int read(ByteBuffer buf) throws IOException {
  if (curDataSlice == null || curDataSlice.remaining() == 0 && bytesNeededToFinish > 0) {
    TraceScope scope = Trace.startSpan(
        "RemoteBlockReader2#readNextPacket(" + blockId + ")", Sampler.NEVER);
    try {
      readNextPacket();
    } finally {
      scope.close();
    }
  }
  if (curDataSlice.remaining() == 0) {
    // we're at EOF now
    return -1;
  }

  int nRead = Math.min(curDataSlice.remaining(), buf.remaining());
  ByteBuffer writeSlice = curDataSlice.duplicate();
  writeSlice.limit(writeSlice.position() + nRead);
  buf.put(writeSlice);
  curDataSlice.position(writeSlice.position());

  return nRead;
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * Reads bytes into a buffer until EOF or the buffer's limit is reached
 */
private int fillBuffer(FileInputStream stream, ByteBuffer buf)
    throws IOException {
  TraceScope scope = Trace.startSpan("BlockReaderLocalLegacy#fillBuffer(" +
      blockId + ")", Sampler.NEVER);
  try {
    int bytesRead = stream.getChannel().read(buf);
    if (bytesRead < 0) {
      //EOF
      return bytesRead;
    }
    while (buf.remaining() > 0) {
      int n = stream.getChannel().read(buf);
      if (n < 0) {
        //EOF
        return bytesRead;
      }
      bytesRead += n;
    }
    return bytesRead;
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * Decrypts a EDEK by consulting the KeyProvider.
 */
private KeyVersion decryptEncryptedDataEncryptionKey(FileEncryptionInfo
    feInfo) throws IOException {
  TraceScope scope = Trace.startSpan("decryptEDEK", traceSampler);
  try {
    KeyProvider provider = getKeyProvider();
    if (provider == null) {
      throw new IOException("No KeyProvider is configured, cannot access" +
          " an encrypted file");
    }
    EncryptedKeyVersion ekv = EncryptedKeyVersion.createForDecryption(
        feInfo.getKeyName(), feInfo.getEzKeyVersionName(), feInfo.getIV(),
        feInfo.getEncryptedDataEncryptionKey());
    try {
      KeyProviderCryptoExtension cryptoProvider = KeyProviderCryptoExtension
          .createKeyProviderCryptoExtension(provider);
      return cryptoProvider.decryptEncryptedKey(ekv);
    } catch (GeneralSecurityException e) {
      throw new IOException(e);
    }
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
public void removeAclEntries(String src, List<AclEntry> aclSpec)
    throws IOException {
  checkOpen();
  TraceScope scope = Trace.startSpan("removeAclEntries", traceSampler);
  try {
    namenode.removeAclEntries(src, aclSpec);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   AclException.class,
                                   FileNotFoundException.class,
                                   NSQuotaExceededException.class,
                                   SafeModeException.class,
                                   SnapshotAccessControlException.class,
                                   UnresolvedPathException.class);
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
public void removeDefaultAcl(String src) throws IOException {
  checkOpen();
  TraceScope scope = Trace.startSpan("removeDefaultAcl", traceSampler);
  try {
    namenode.removeDefaultAcl(src);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   AclException.class,
                                   FileNotFoundException.class,
                                   NSQuotaExceededException.class,
                                   SafeModeException.class,
                                   SnapshotAccessControlException.class,
                                   UnresolvedPathException.class);
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
public void removeAcl(String src) throws IOException {
  checkOpen();
  TraceScope scope = Trace.startSpan("removeAcl", traceSampler);
  try {
    namenode.removeAcl(src);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   AclException.class,
                                   FileNotFoundException.class,
                                   NSQuotaExceededException.class,
                                   SafeModeException.class,
                                   SnapshotAccessControlException.class,
                                   UnresolvedPathException.class);
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
public void setAcl(String src, List<AclEntry> aclSpec) throws IOException {
  checkOpen();
  TraceScope scope = Trace.startSpan("setAcl", traceSampler);
  try {
    namenode.setAcl(src, aclSpec);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   AclException.class,
                                   FileNotFoundException.class,
                                   NSQuotaExceededException.class,
                                   SafeModeException.class,
                                   SnapshotAccessControlException.class,
                                   UnresolvedPathException.class);
  } finally {
    scope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * Returns the next batch of events in the stream, waiting indefinitely if
 * a new batch  is not immediately available.
 *
 * @throws IOException see {@link DFSInotifyEventInputStream#poll()}
 * @throws MissingEventsException see
 * {@link DFSInotifyEventInputStream#poll()}
 * @throws InterruptedException if the calling thread is interrupted
 */
public EventBatch take() throws IOException, InterruptedException,
    MissingEventsException {
  TraceScope scope = Trace.startSpan("inotifyTake", traceSampler);
  EventBatch next = null;
  try {
    int nextWaitMin = INITIAL_WAIT_MS;
    while ((next = poll()) == null) {
      // sleep for a random period between nextWaitMin and nextWaitMin * 2
      // to avoid stampedes at the NN if there are multiple clients
      int sleepTime = nextWaitMin + rng.nextInt(nextWaitMin);
      LOG.debug("take(): poll() returned null, sleeping for {} ms", sleepTime);
      Thread.sleep(sleepTime);
      // the maximum sleep is 2 minutes
      nextWaitMin = Math.min(60000, nextWaitMin * 2);
    }
  } finally {
    scope.close();
  }

  return next;
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@Override
public HdfsBlocksMetadata call() throws Exception {
  HdfsBlocksMetadata metadata = null;
  // Create the RPC proxy and make the RPC
  ClientDatanodeProtocol cdp = null;
  TraceScope scope =
      Trace.startSpan("getHdfsBlocksMetadata", parentSpan);
  try {
    cdp = DFSUtil.createClientDatanodeProtocolProxy(datanode, configuration,
        timeout, connectToDnViaHostname);
    metadata = cdp.getHdfsBlocksMetadata(poolId, blockIds, dnTokens);
  } catch (IOException e) {
    // Bubble this up to the caller, handle with the Future
    throw e;
  } finally {
    scope.close();
    if (cdp != null) {
      RPC.stopProxy(cdp);
    }
  }
  return metadata;
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
public static RpcRequestHeaderProto makeRpcRequestHeader(RPC.RpcKind rpcKind,
    RpcRequestHeaderProto.OperationProto operation, int callId,
    int retryCount, byte[] uuid) {
  RpcRequestHeaderProto.Builder result = RpcRequestHeaderProto.newBuilder();
  result.setRpcKind(convert(rpcKind)).setRpcOp(operation).setCallId(callId)
      .setRetryCount(retryCount).setClientId(ByteString.copyFrom(uuid));

  // Add tracing info if we are currently tracing.
  if (Trace.isTracing()) {
    Span s = Trace.currentSpan();
    result.setTraceInfo(RPCTraceInfoProto.newBuilder()
        .setParentId(s.getSpanId())
        .setTraceId(s.getTraceId()).build());
  }

  return result.build();
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@Override
public Object invoke(Object proxy, Method method, Object[] args)
  throws Throwable {
  long startTime = 0;
  if (LOG.isDebugEnabled()) {
    startTime = Time.now();
  }
  TraceScope traceScope = null;
  if (Trace.isTracing()) {
    traceScope = Trace.startSpan(RpcClientUtil.methodToTraceString(method));
  }
  ObjectWritable value;
  try {
    value = (ObjectWritable)
      client.call(RPC.RpcKind.RPC_WRITABLE, new Invocation(method, args),
        remoteId, fallbackToSimpleAuth);
  } finally {
    if (traceScope != null) traceScope.close();
  }
  if (LOG.isDebugEnabled()) {
    long callTime = Time.now() - startTime;
    LOG.debug("Call: " + method.getName() + " " + callTime);
  }
  return value.get();
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@Override
protected void populatePool(ExecutorService pool) {
  RegionStates regionStates = assignmentManager.getRegionStates();
  for (HRegionInfo region: regions) {
    if (regionStates.isRegionInTransition(region)
        && !regionStates.isRegionInState(region, State.FAILED_CLOSE)) {
      continue;
    }
    final HRegionInfo hri = region;
    pool.execute(Trace.wrap("DisableTableHandler.BulkDisabler",new Runnable() {
      public void run() {
        assignmentManager.unassign(hri, true);
      }
    }));
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@Override
protected void populatePool(ExecutorService pool) {
  RegionStates regionStates = assignmentManager.getRegionStates();
  for (final HRegionInfo region : regions) {
    if (regionStates.isRegionInTransition(region)
        && !regionStates.isRegionInState(region, RegionState.State.FAILED_CLOSE)) {
      continue;
    }
    pool.execute(Trace.wrap("DisableTableHandler.BulkDisabler", new Runnable() {
      @Override
      public void run() {
        assignmentManager.unassign(region);
      }
    }));
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NP_NULL_ON_SOME_PATH_EXCEPTION",
    justification="Will never be null")
@Override
public long append(final HTableDescriptor htd, final HRegionInfo hri, final WALKey key,
    final WALEdit edits, final boolean inMemstore) throws IOException {
  if (this.closed) throw new IOException("Cannot append; log is closed");
  // Make a trace scope for the append.  It is closed on other side of the ring buffer by the
  // single consuming thread.  Don't have to worry about it.
  TraceScope scope = Trace.startSpan("FSHLog.append");

  // This is crazy how much it takes to make an edit.  Do we need all this stuff!!!!????  We need
  // all this to make a key and then below to append the edit, we need to carry htd, info,
  // etc. all over the ring buffer.
  FSWALEntry entry = null;
  long sequence = this.disruptor.getRingBuffer().next();
  try {
    RingBufferTruck truck = this.disruptor.getRingBuffer().get(sequence);
    // TODO: reuse FSWALEntry as we do SyncFuture rather create per append.
    entry = new FSWALEntry(sequence, key, edits, htd, hri, inMemstore);
    truck.loadPayload(entry, scope.detach());
  } finally {
    this.disruptor.getRingBuffer().publish(sequence);
  }
  return sequence;
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
MemStoreScanner(long readPoint) {
  super();

  this.readPoint = readPoint;
  cellSetAtCreation = cellSet;
  snapshotAtCreation = snapshot;
  if (allocator != null) {
    this.allocatorAtCreation = allocator;
    this.allocatorAtCreation.incScannerCount();
  }
  if (snapshotAllocator != null) {
    this.snapshotAllocatorAtCreation = snapshotAllocator;
    this.snapshotAllocatorAtCreation.incScannerCount();
  }
  if (Trace.isTracing() && Trace.currentSpan() != null) {
    Trace.currentSpan().addTimelineAnnotation("Creating MemStoreScanner");
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * Provides an extension point for tests that don't want a per row invocation.
 */
void testTimed() throws IOException, InterruptedException {
  int startRow = getStartRow();
  int lastRow = getLastRow();
  // Report on completion of 1/10th of total.
  for (int i = startRow; i < lastRow; i++) {
    if (i % everyN != 0) continue;
    long startTime = System.nanoTime();
    TraceScope scope = Trace.startSpan("test row", traceSampler);
    try {
      testRow(i);
    } finally {
      scope.close();
    }
    latency.update((System.nanoTime() - startTime) / 1000);
    if (status != null && i > 0 && (i % getReportingPeriod()) == 0) {
      status.setStatus(generateStatus(startRow, i, lastRow));
    }
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * <p>
 * NONSEQUENTIAL create is idempotent operation.
 * Retry before throwing exceptions.
 * But this function will not throw the NodeExist exception back to the
 * application.
 * </p>
 * <p>
 * But SEQUENTIAL is NOT idempotent operation. It is necessary to add
 * identifier to the path to verify, whether the previous one is successful
 * or not.
 * </p>
 *
 * @return Path
 */
public String create(String path, byte[] data, List<ACL> acl,
    CreateMode createMode)
throws KeeperException, InterruptedException {
  TraceScope traceScope = null;
  try {
    traceScope = Trace.startSpan("RecoverableZookeeper.create");
    byte[] newData = appendMetaData(data);
    switch (createMode) {
      case EPHEMERAL:
      case PERSISTENT:
        return createNonSequential(path, newData, acl, createMode);

      case EPHEMERAL_SEQUENTIAL:
      case PERSISTENT_SEQUENTIAL:
        return createSequential(path, newData, acl, createMode);

      default:
        throw new IllegalArgumentException("Unrecognized CreateMode: " +
            createMode);
    }
  } finally {
    if (traceScope != null) traceScope.close();
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
/**
 * Reads the names of classes specified in the {@code hbase.trace.spanreceiver.classes} property
 * and instantiates and registers them with the Tracer.
 *
 */
public void loadSpanReceivers() {
  String[] receiverNames = conf.getStrings(SPAN_RECEIVERS_CONF_KEY);
  if (receiverNames == null || receiverNames.length == 0) {
    return;
  }

  SpanReceiverBuilder builder = new SpanReceiverBuilder(new HBaseHTraceConfiguration(conf));
  for (String className : receiverNames) {
    className = className.trim();

    SpanReceiver receiver = builder.spanReceiverClass(className).build();
    if (receiver != null) {
      receivers.add(receiver);
      LOG.info("SpanReceiver " + className + " was loaded successfully.");
    }
  }
  for (SpanReceiver rcvr : receivers) {
    Trace.addReceiver(rcvr);
  }
}
 

import org.apache.htrace.Trace; //导入依赖的package包/类
private void createEntries(Opts opts) throws TableNotFoundException, AccumuloException, AccumuloSecurityException {

    // Trace the write operation. Note, unless you flush the BatchWriter, you will not capture
    // the write operation as it is occurs asynchronously. You can optionally create additional Spans
    // within a given Trace as seen below around the flush
    TraceScope scope = Trace.startSpan("Client Write", Sampler.ALWAYS);

    System.out.println("TraceID: " + Long.toHexString(scope.getSpan().getTraceId()));
    BatchWriter batchWriter = opts.getConnector().createBatchWriter(opts.getTableName(), new BatchWriterConfig());

    Mutation m = new Mutation("row");
    m.put("cf", "cq", "value");

    batchWriter.addMutation(m);
    // You can add timeline annotations to Spans which will be able to be viewed in the Monitor
    scope.getSpan().addTimelineAnnotation("Initiating Flush");
    batchWriter.flush();

    batchWriter.close();
    scope.close();
  }