fix: fix batcher metric labels (#1829)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/java-bigtable/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://togithub.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: mutianf

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -145,4 +145,9 @@
         <className>com/google/cloud/bigtable/data/v2/stub/readrows/RowMerger</className>
         <method>*</method>
     </difference>
+    <!-- InternalApi was removed -->
+    <difference>
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/metrics/BigtableTracerBatchedUnaryCallable</className>
+    </difference>
 </differences>

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

@@ -87,7 +87,6 @@
 import com.google.cloud.bigtable.data.v2.stub.changestream.GenerateInitialChangeStreamPartitionsUserCallable;
 import com.google.cloud.bigtable.data.v2.stub.changestream.ReadChangeStreamResumptionStrategy;
 import com.google.cloud.bigtable.data.v2.stub.changestream.ReadChangeStreamUserCallable;
-import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerBatchedUnaryCallable;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerStreamingCallable;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerUnaryCallable;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTracerFactory;
@@ -509,11 +508,8 @@ private <RowT> UnaryCallable<Query, List<RowT>> createBulkReadRowsCallable(
     UnaryCallable<Query, List<RowT>> tracedBatcher =
         new TracedBatcherUnaryCallable<>(readRowsUserCallable.all());
 
-    UnaryCallable<Query, List<RowT>> withBigtableTracer =
-        new BigtableTracerBatchedUnaryCallable<>(tracedBatcher);
-
     UnaryCallable<Query, List<RowT>> traced =
-        new TracedUnaryCallable<>(withBigtableTracer, clientContext.getTracerFactory(), span);
+        new TracedUnaryCallable<>(tracedBatcher, clientContext.getTracerFactory(), span);
 
     return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
   }
@@ -641,10 +637,9 @@ private UnaryCallable<BulkMutation, Void> createBulkMutateRowsCallable() {
     UnaryCallable<BulkMutation, Void> tracedBatcherUnaryCallable =
         new TracedBatcherUnaryCallable<>(userFacing);
 
-    UnaryCallable<BulkMutation, Void> withBigtableTracer =
-        new BigtableTracerBatchedUnaryCallable<>(tracedBatcherUnaryCallable);
     UnaryCallable<BulkMutation, Void> traced =
-        new TracedUnaryCallable<>(withBigtableTracer, clientContext.getTracerFactory(), spanName);
+        new TracedUnaryCallable<>(
+            tracedBatcherUnaryCallable, clientContext.getTracerFactory(), spanName);
 
     return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
   }
@@ -747,6 +742,9 @@ public Map<String, String> extract(MutateRowsRequest mutateRowsRequest) {
     ServerStreamingCallable<MutateRowsRequest, MutateRowsResponse> convertException =
         new ConvertExceptionCallable<>(callable);
 
+    ServerStreamingCallable<MutateRowsRequest, MutateRowsResponse> withBigtableTracer =
+        new BigtableTracerStreamingCallable<>(convertException);
+
     RetryAlgorithm<Void> retryAlgorithm =
         new RetryAlgorithm<>(
             new ApiResultRetryAlgorithm<Void>(),
@@ -757,7 +755,7 @@ public Map<String, String> extract(MutateRowsRequest mutateRowsRequest) {
 
     return new MutateRowsRetryingCallable(
         clientContext.getDefaultCallContext(),
-        convertException,
+        withBigtableTracer,
         retryingExecutor,
         settings.bulkMutateRowsSettings().getRetryableCodes());
   }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableTracerBatchedUnaryCallable.java

@@ -70,7 +70,7 @@ public ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context)
     }
   }
 
-  class BigtableTracerUnaryCallback<ResponseT> implements ApiFutureCallback<ResponseT> {
+  private class BigtableTracerUnaryCallback<ResponseT> implements ApiFutureCallback<ResponseT> {
 
     private final BigtableTracer tracer;
     private final GrpcResponseMetadata responseMetadata;

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableTracerUnaryCallable.java

@@ -491,12 +491,16 @@ public void testBatchBlockingLatencies() throws InterruptedException {
       verify(statsRecorderWrapper, timeout(1000).times(expectedNumRequests))
           .putClientBlockingLatencies(throttledTime.capture());
 
-      // Adding the first 2 elements should not get throttled since the batch is empty
-      assertThat(throttledTime.getAllValues().get(0)).isEqualTo(0);
       // After the first request is sent, batcher will block on add because of the server latency.
       // Blocking latency should be around server latency.
       assertThat(throttledTime.getAllValues().get(1)).isAtLeast(SERVER_LATENCY - 10);
       assertThat(throttledTime.getAllValues().get(2)).isAtLeast(SERVER_LATENCY - 10);
+
+      verify(statsRecorderWrapper, timeout(100).times(expectedNumRequests))
+          .recordAttempt(status.capture(), tableId.capture(), zone.capture(), cluster.capture());
+
+      assertThat(zone.getAllValues()).containsExactly(ZONE, ZONE, ZONE);
+      assertThat(cluster.getAllValues()).containsExactly(CLUSTER, CLUSTER, CLUSTER);
     }
   }