feat: enable last_scanned_row_responses feature flag (#1862)

This enables the last_scanned_row_responses feature flag and updates the unit test for scan markers to have slightly better coverage.

Author: steveniemitz

README.md

@@ -50,7 +50,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.20.0')
+implementation platform('com.google.cloud:libraries-bom:26.21.0')
 
 implementation 'com.google.cloud:google-cloud-bigtable'
 ```

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

@@ -731,7 +731,8 @@ private Builder() {
               .setTotalTimeout(PRIME_REQUEST_TIMEOUT)
               .build());
 
-      featureFlags = FeatureFlags.newBuilder().setReverseScans(true);
+      featureFlags =
+          FeatureFlags.newBuilder().setReverseScans(true).setLastScannedRowResponses(true);
     }
 
     private Builder(EnhancedBigtableStubSettings settings) {

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

@@ -243,6 +243,7 @@ public void testFeatureFlags() throws InterruptedException, IOException, Executi
         FeatureFlags.parseFrom(BaseEncoding.base64Url().decode(encodedFeatureFlags));
 
     assertThat(featureFlags.getReverseScans()).isTrue();
+    assertThat(featureFlags.getLastScannedRowResponses()).isTrue();
   }
 
   @Test

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/RowMergingCallableTest.java

@@ -42,9 +42,18 @@
 public class RowMergingCallableTest {
   @Test
   public void scanMarker() {
+    ReadRowsResponse.Builder rrr = ReadRowsResponse.newBuilder();
+    rrr.addChunksBuilder()
+        .setRowKey(ByteString.copyFromUtf8("key0"))
+        .setFamilyName(StringValue.of("f1"))
+        .setQualifier(BytesValue.of(ByteString.copyFromUtf8("q1")))
+        .setCommitRow(true);
+
     FakeStreamingApi.ServerStreamingStashCallable<ReadRowsRequest, ReadRowsResponse> inner =
         new ServerStreamingStashCallable<>(
             Lists.newArrayList(
+                // send a row
+                rrr.build(),
                 // send a scan marker
                 ReadRowsResponse.newBuilder()
                     .setLastScannedRowKey(ByteString.copyFromUtf8("key1"))
@@ -56,6 +65,15 @@ public void scanMarker() {
 
     Truth.assertThat(results)
         .containsExactly(
+            Row.create(
+                ByteString.copyFromUtf8("key0"),
+                Lists.newArrayList(
+                    RowCell.create(
+                        "f1",
+                        ByteString.copyFromUtf8("q1"),
+                        0,
+                        Lists.newArrayList(),
+                        ByteString.EMPTY))),
             Row.create(ByteString.copyFromUtf8("key1"), Lists.<RowCell>newArrayList()));
   }