fix: prevent scanner cancellation from reading an extra row (#3759)

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://github.com/googleapis/java-bigtable-hbase/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://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: igorbernstein2

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/veneer/DataClientVeneerApi.java

@@ -254,9 +254,7 @@ public Result next() {
 
     @Override
     public void close() {
-      if (iterator.hasNext()) {
-        serverStream.cancel();
-      }
+      serverStream.cancel();
     }
 
     public boolean renewLease() {

bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/wrappers/veneer/TestDataClientVeneerApi.java

@@ -48,6 +48,7 @@
 import com.google.protobuf.ByteString;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.List;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.client.Result;
@@ -224,6 +225,34 @@ public void testReadRows() throws IOException {
         .call(Mockito.eq(query), Mockito.any(GrpcCallContext.class));
   }
 
+  @Test
+  public void testReadRowsCancel() throws IOException {
+
+    Query query = Query.create(TABLE_ID).rowKey(ROW_KEY);
+    when(mockDataClient.readRowsCallable(Mockito.<RowResultAdapter>any()))
+        .thenReturn(mockStreamingCallable)
+        .thenReturn(mockStreamingCallable);
+
+    when(mockStreamingCallable.call(Mockito.eq(query), Mockito.any(GrpcCallContext.class)))
+        .thenReturn(serverStream);
+
+    Iterator<Result> mockIter = Mockito.mock(Iterator.class);
+    when(serverStream.iterator()).thenReturn(mockIter);
+    when(mockIter.hasNext()).thenReturn(true);
+    when(mockIter.next()).thenReturn(EXPECTED_RESULT);
+
+    ResultScanner resultScanner = dataClientWrapper.readRows(query);
+    assertResult(EXPECTED_RESULT, resultScanner.next());
+
+    doNothing().when(serverStream).cancel();
+    resultScanner.close();
+
+    // make sure that the scanner doesn't iteract with the iterator on close
+    verify(serverStream).cancel();
+    verify(mockIter, times(1)).hasNext();
+    verify(mockIter, times(1)).next();
+  }
+
   @Test
   public void testReadRowsAsync() throws Exception {
     Query query = Query.create(TABLE_ID).rowKey(ROW_KEY);

bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/test/java/com/google/cloud/bigtable/hbase1_x/TestMetrics.java

@@ -271,9 +271,7 @@ public void testFirstResponseLatency() throws IOException {
             .get("google-cloud-bigtable.grpc.method.ReadRows.firstResponse.latency")
             .get();
 
-    // adding buffer time to the upper range to allow for a race between the emulator and the client
-    // recording the duration
-    assertThat(firstResponseLatencyMetric).isAtMost(methodInvocationLatency - 20 / 2);
+    assertThat(firstResponseLatencyMetric).isAtMost(methodInvocationLatency);
   }
 
   @Test