fix: fix MutateRowsAttemptCallable to avoid NPE in MetricTracer (#557)

Author: mutianf

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/mutaterows/MutateRowsAttemptCallable.java

@@ -160,9 +160,17 @@ public void setExternalFuture(RetryingFuture<Void> externalFuture) {
   @Override
   public Void call() {
     try {
+      // externalFuture is set from MutateRowsRetryingCallable before invoking this method. It
+      // shouldn't be null unless the code changed
       Preconditions.checkNotNull(
           externalFuture, "External future must be set before starting an attempt");
 
+      // attemptStared should be called at the very start of the operation. This will initialize
+      // variables in ApiTracer and avoid exceptions when the tracer marks the attempt as finished
+      callContext
+          .getTracer()
+          .attemptStarted(externalFuture.getAttemptSettings().getOverallAttemptCount());
+
       Preconditions.checkState(
           currentRequest.getEntriesCount() > 0, "Request doesn't have any mutations to send");
 
@@ -179,10 +187,6 @@ public Void call() {
         return null;
       }
 
-      callContext
-          .getTracer()
-          .attemptStarted(externalFuture.getAttemptSettings().getOverallAttemptCount());
-
       // Make the actual call
       ApiFuture<List<MutateRowsResponse>> innerFuture =
           innerCallable.futureCall(currentRequest, currentCallContext);

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/MetricsTracerTest.java

@@ -26,6 +26,7 @@
 import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
 import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
 import com.google.cloud.bigtable.data.v2.FakeServiceHelper;
+import com.google.cloud.bigtable.data.v2.models.BulkMutation;
 import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
@@ -46,6 +47,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -327,6 +329,24 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
     assertThat(attemptLatency).isIn(Range.closed(sleepTime, elapsed - sleepTime));
   }
 
+  @Test
+  public void testInvalidRequest() throws InterruptedException {
+    try {
+      stub.bulkMutateRowsCallable().call(BulkMutation.create(TABLE_ID));
+      Assert.fail("Invalid request should throw exception");
+    } catch (IllegalStateException e) {
+      Thread.sleep(100);
+      // Verify that the latency is recorded with an error code (in this case UNKNOWN)
+      long attemptLatency =
+          getAggregationValueAsLong(
+              RpcViewConstants.BIGTABLE_ATTEMPT_LATENCY_VIEW,
+              ImmutableMap.of(
+                  RpcMeasureConstants.BIGTABLE_OP, TagValue.create("Bigtable.MutateRows"),
+                  RpcMeasureConstants.BIGTABLE_STATUS, TagValue.create("UNKNOWN")));
+      assertThat(attemptLatency).isAtLeast(0);
+    }
+  }
+
   @SuppressWarnings("unchecked")
   private static <T> StreamObserver<T> anyObserver(Class<T> returnType) {
     return (StreamObserver<T>) any(returnType);