fix(bigtable): fix race condition between vrpc cancel and start

Author: mutianf

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/VRpcImpl.java

@@ -72,6 +72,7 @@ private enum State {
   private PeerInfo peerInfo;
 
   private AtomicReference<State> state;
+  private AtomicReference<Status> cancelStatus = new AtomicReference<>();
 
   private final DebugTagTracer debugTagTracer;
 
@@ -97,7 +98,12 @@ public void start(ReqT req, VRpcCallContext ctx, VRpcListener<RespT> listener) {
     boolean retryable = true;
 
     if (!state.compareAndSet(State.NEW, State.STARTED)) {
-      status = Status.INTERNAL.withDescription("VRpc already started in state: " + state.get());
+      // The vRPC was cancelled before it's started
+      status = cancelStatus.get();
+      if (status == null) {
+        // Otherwise there's a double-start bug
+        status = Status.INTERNAL.withDescription("VRpc already started in state: " + state.get());
+      }
       retryable = false;
     } else if (ctx.getOperationInfo().getDeadline().timeRemaining(TimeUnit.MICROSECONDS)
         < TimeUnit.MILLISECONDS.toMicros(1)) {
@@ -195,6 +201,19 @@ void handleError(VRpcResult result) {
 
   @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
+    // Try to transition from NEW to CLOSED to capture pre-start cancellations
+    if (state.compareAndSet(State.NEW, State.CLOSED)) {
+      Status status = Status.CANCELLED;
+      if (message != null) {
+        status = status.withDescription(message);
+      }
+      if (cause != null) {
+        status = status.withCause(cause);
+      }
+      cancelStatus.set(status);
+      return;
+    }
+    // If the vRPC is already started, delegate the cancel to session
     session.cancelRpc(rpcId, message, cause);
   }