code: protected void run() {
      try (Scope scope = TRACER.withSpan(operationSpan)) {
        rpcTimerContext = rpc.getRpcMetrics().timeRpc();
        operationSpan.addAnnotation(Annotation.fromDescriptionAndAttributes("rpcStart",
          ImmutableMap.of("attempt", AttributeValue.longAttributeValue(failedCount))));
        Metadata metadata = new Metadata();
        metadata.merge(originalMetadata);
        callWrapper.setCallAndStart(rpc, getRpcCallOptions(), getRetryRequest(), this, metadata);
      } catch (Exception e) {
        setException(e);
      }
    }
docs: Calls {@link BigtableAsyncRpc#newCall(CallOptions)} and
  {@link BigtableAsyncRpc#start(Object, io.grpc.ClientCall.Listener, Metadata, ClientCall)} }
  with this as the listener so that retries happen correctly.