Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1651 (2018-02-06 16:26:50)

Removing an NPE when auth token retrieval fails fast
`AbstractRetryingOperation` now uses a no-op Call to replace the notion of "null".

When Auth fails, the interceptor immediately calls close, thereby setting call to null. `RetryingReadRowsOperation` broke in those conditions.  I added a test to confirm this behavior before the fix, and then retested with `AbstractRetryingOperation`.


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
index de312810d..786b1d50f 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
@@ -57,9 +57,34 @@ import io.opencensus.trace.Tracing;
 /**
  * A {@link ClientCall.Listener} that retries a {@link BigtableAsyncRpc} request.
  */
+@SuppressWarnings("unchecked")
 public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
     extends ClientCall.Listener<ResponseT>  {
 
+  @SuppressWarnings("rawtypes")
+  private static final ClientCall NULL_CALL = new ClientCall() {
+
+    @Override
+    public void start(Listener responseListener, Metadata headers) {
+    }
+
+    @Override
+    public void request(int numMessages) {
+    }
+
+    @Override
+    public void cancel(String message, Throwable cause) {
+    }
+
+    @Override
+    public void halfClose() {
+    }
+
+    @Override
+    public void sendMessage(Object message) {
+    }
+  };
+
   /** Constant <code>LOG</code> */
   protected static final Logger LOG = new Logger(AbstractRetryingOperation.class);
   private static final Tracer TRACER = Tracing.getTracer();
@@ -111,7 +136,7 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
 
   protected final GrpcFuture<ResultT> completionFuture;
   protected Object callLock = new String("");
-  protected ClientCall<RequestT, ResponseT> call;
+  protected ClientCall<RequestT, ResponseT> call = NULL_CALL;
   protected Timer.Context operationTimerContext;
   protected Timer.Context rpcTimerContext;
 
@@ -150,7 +175,7 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
   public void onClose(Status status, Metadata trailers) {
     try (Closeable s = TRACER.withSpan(operationSpan)) {
       synchronized (callLock) {
-        call = null;
+        call = NULL_CALL;
       }
       rpcTimerContext.close();
       // OK
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
index 12c81c572..63a661637 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
@@ -32,7 +32,6 @@ import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.ByteString;
 
 import io.grpc.CallOptions;
-import io.grpc.ClientCall;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.stub.ClientCallStreamObserver;
@@ -53,16 +52,11 @@ public class RetryingReadRowsOperation extends
 
   private static final String TIMEOUT_CANCEL_MSG = "Client side timeout induced cancellation";
 
-  private static class CallToStreamObserverAdapter<T> extends ClientCallStreamObserver<T> {
-    private final ClientCall<T, ?> call;
+  private class CallToStreamObserverAdapter extends ClientCallStreamObserver<ReadRowsRequest> {
     private boolean autoFlowControlEnabled = true;
 
-    public CallToStreamObserverAdapter(ClientCall<T, ?> call) {
-      this.call = call;
-    }
-
     @Override
-    public void onNext(T value) {
+    public void onNext(ReadRowsRequest value) {
       call.sendMessage(value);
     }
 
@@ -115,7 +109,7 @@ public class RetryingReadRowsOperation extends
 
   // The number of times we've retried after a timeout
   private int timeoutRetryCount = 0;
-  private CallToStreamObserverAdapter<ReadRowsRequest> adapter;
+  private CallToStreamObserverAdapter adapter;
   private StreamObserver<ReadRowsResponse> resultObserver;
   private int totalRowsProcessed = 0;
   private volatile ReadRowsRequest nextRequest;
@@ -151,16 +145,16 @@ public class RetryingReadRowsOperation extends
   @Override
   public void run() {
     // restart the clock.
-    lastResponseMs = clock.currentTimeMillis();
     this.rowMerger = new RowMerger(rowObserver);
+    adapter = new CallToStreamObserverAdapter();
     synchronized (callLock) {
       super.run();
       // pre-fetch one more result, for performance reasons.
-      adapter = new CallToStreamObserverAdapter<ReadRowsRequest>(call);
       adapter.request(1);
       if (rowObserver instanceof ClientResponseObserver) {
         ((ClientResponseObserver<ReadRowsRequest, FlatRow>) rowObserver).beforeStart(adapter);
       }
+      lastResponseMs = clock.currentTimeMillis();
     }
   }
 


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
index 1286338eb..359b45274 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
@@ -37,6 +37,7 @@ import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -336,6 +337,25 @@ public class RetryingReadRowsOperationTest {
     performTimeout(underTest, time);
   }
 
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testImmediateOnClose() {
+    Mockito.doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) throws Throwable {
+        invocation.getArgumentAt(1, ClientCall.Listener.class).onClose(Status.OK, new Metadata());
+        return null;
+      }
+    }).when(mockRetryableRpc).start(any(ReadRowsRequest.class), any(ClientCall.Listener.class),
+      any(Metadata.class), eq(mockClientCall));
+
+    RetryingReadRowsOperation underTest = createOperation(mockFlatRowObserver);
+
+    // The test revolves around this call not throwing an exception.  It did at one point with
+    // an invocation of call.request(1) when call is null.
+    start(underTest);
+  }
+
   protected void performTimeout(RetryingReadRowsOperation underTest, AtomicLong time)
       throws BigtableRetriesExhaustedException {
     time.addAndGet(RETRY_OPTIONS.getReadPartialRowTimeoutMillis() + 1);