Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1938 (2018-09-13 13:05:00)

Fixing some async methods


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
index 0f84e2cc7..d77e69068 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
@@ -91,6 +91,15 @@ class ReadRowsRequestManager {
     }
     RowSet.Builder rowSetBuilder = RowSet.newBuilder();
 
+    if (originalRows.getRowKeysCount() == 0 && originalRows.getRowRangesCount() == 0) {
+      rowSetBuilder.addRowRanges(
+          RowRange
+              .newBuilder()
+              .setStartKeyOpen(lastFoundKey)
+              .setEndKeyOpen(ByteString.EMPTY)
+              .build());
+    }
+
     for (ByteString key : originalRows.getRowKeysList()) {
       if (!startKeyIsAlreadyRead(key)) {
         rowSetBuilder.addRowKeys(key);
@@ -107,16 +116,16 @@ class ReadRowsRequestManager {
         continue;
       }
 
-      RowRange newRange = rowRange;
+      RowRange newRowRange = rowRange;
       StartKeyCase startKeyCase = rowRange.getStartKeyCase();
       if ((startKeyCase == StartKeyCase.START_KEY_CLOSED
           && startKeyIsAlreadyRead(rowRange.getStartKeyClosed()))
           || (startKeyCase == StartKeyCase.START_KEY_OPEN
           && startKeyIsAlreadyRead(rowRange.getStartKeyOpen()))
           || startKeyCase == StartKeyCase.STARTKEY_NOT_SET) {
-        newRange = rowRange.toBuilder().setStartKeyOpen(lastFoundKey).build();
+        newRowRange = newRowRange.toBuilder().setStartKeyOpen(lastFoundKey).build();
       }
-      rowSetBuilder.addRowRanges(newRange);
+      rowSetBuilder.addRowRanges(newRowRange);
     }
 
     return rowSetBuilder.build();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
index adcb6e0ec..afff1c392 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.TreeMap;
 
 import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.util.ByteStringComparator;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
 import com.google.bigtable.v2.ReadRowsRequest;
@@ -121,14 +122,16 @@ protected static final Logger LOG = new Logger(RowMerger.class);
         final ByteString rowKey = newChunk.getRowKey();
         Preconditions.checkArgument(!rowKey.isEmpty(), "A row key must be set: %s",
           newChunk);
-        Preconditions.checkState(previousKey == null || !rowKey.equals(previousKey),
-          "A commit happened but the same key followed: %s", newChunk);
+        if (previousKey != null &&
+            ByteStringComparator.INSTANCE.compare(previousKey, rowKey) >= 0) {
+          throw new IllegalArgumentException(String
+              .format("Found key '%s' after key '%s'", rowKey.toStringUtf8(),
+                  previousKey.toStringUtf8()));
+        }
         Preconditions.checkArgument(newChunk.hasQualifier(), "A column qualifier must be set: %s",
           newChunk);
-        if (newChunk.getValueSize() > 0) {
-          Preconditions.checkArgument(!newChunk.getCommitRow(),
-            "A row cannot be have a value size and be a commit row: %s", newChunk);
-        }
+        Preconditions.checkArgument(!newChunk.getCommitRow() || newChunk.getValueSize() == 0,
+          "A row cannot be have a value size and be a commit row: %s", newChunk);
       }
 
       @Override
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
index f8bf092f9..feade1902 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
@@ -343,7 +343,7 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
   @Override
   public List<SnapshotDescription> listTableSnapshots(Pattern tableName,
       Pattern snapshotName) throws IOException {
-    List<SnapshotDescription> response = new ArrayList<SnapshotDescription>();
+    List<SnapshotDescription> response = new ArrayList<>();
     for (SnapshotDescription snapshot: listSnapshots(snapshotName)) {
       if (tableName.matcher(snapshot.getTableNameAsString()).matches()) {
         response.add(snapshot);
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
index a918d70f0..9aa5dd988 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
@@ -511,7 +511,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
         bigtableTableAdminClient.listSnapshotsAsync(request)
             .thenApply(r -> r.getSnapshotsList()
                 .stream()
-                .map(BigtableAsyncAdmin::toSnapshotDesscription)
+                .map(BigtableAsyncAdmin::toSnapshotDescription)
                 .collect(Collectors.toList())
             )
     );
@@ -520,18 +520,19 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   @Override
   public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern pattern) {
     return listSnapshots().thenApply(r ->
-        filter(r, d -> pattern.matcher(d.getName()).matches()));
+        filter(r, d -> pattern == null || pattern.matcher(d.getName()).matches()));
   }
 
   @Override
   public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
-      Pattern pattern) {
-    return listSnapshots(pattern).thenApply(r ->
-       filter(r, d -> pattern.matcher(d.getTableNameAsString()).matches())
+      Pattern snapshotPattern) {
+    return listSnapshots(snapshotPattern).thenApply(r ->
+       filter(r, d -> tableNamePattern == null ||
+           tableNamePattern.matcher(d.getTableNameAsString()).matches())
     );
   }
 
-  private static SnapshotDescription toSnapshotDesscription(Snapshot snapshot) {
+  private static SnapshotDescription toSnapshotDescription(Snapshot snapshot) {
     return new SnapshotDescription(
         snapshot.getName(),
         TableName.valueOf(snapshot.getSourceTable().getName()));
@@ -597,6 +598,16 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
         );
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public CompletableFuture<Void> addColumnFamily(TableName tableName, ColumnFamilyDescriptor columnFamilyDesc) {
+    return modifyColumns(tableName, Modification
+        .newBuilder()
+        .setId(columnFamilyDesc.getNameAsString())
+        .setCreate(tableAdapter2x.toColumnFamily(columnFamilyDesc))
+        .build());
+  }
+
   private SampledRowKeysAdapter getSampledRowKeysAdapter(TableName tableNameAdapter,
       ServerName serverNameAdapter) {
     return new SampledRowKeysAdapter(tableNameAdapter, serverNameAdapter) {
@@ -610,6 +621,48 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     };
   }
 
+  private BigtableClusterName getSnapshotClusterName() throws IOException {
+    if (bigtableSnapshotClusterName == null) {
+      try {
+        bigtableSnapshotClusterName = getClusterName();
+      } catch (IllegalStateException e) {
+        throw new IllegalStateException(
+            "Failed to determine which cluster to use for snapshots, please configure it using "
+                + BigtableOptionsFactory.BIGTABLE_SNAPSHOT_CLUSTER_ID_KEY);
+      }
+    }
+    return bigtableSnapshotClusterName;
+  }
+
+  /**
+   * <p>modifyColumns.</p>
+   *
+   * @param tableName a {@link org.apache.hadoop.hbase.TableName} object.
+   * @param modification an array of {@link com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest.Modification} objectss
+   */
+  private CompletableFuture<Void> modifyColumns(TableName tableName, Modification... modifications) {
+    ModifyColumnFamiliesRequest request = ModifyColumnFamiliesRequest
+        .newBuilder()
+        .addAllModifications(Arrays.asList(modifications))
+        .setName(toBigtableName(tableName))
+        .build();
+    return bigtableTableAdminClient.modifyColumnFamilyAsync(request).thenApply(r -> null);
+  }
+
+  private BigtableClusterName getClusterName() throws IOException {
+    return asyncConnection.getSession().getClusterName();
+  }
+
+  /**
+   * <p>toBigtableName.</p>
+   *
+   * @param tableName a {@link org.apache.hadoop.hbase.TableName} object.
+   * @return a {@link java.lang.String} object.
+   */
+  private String toBigtableName(TableName tableName) {
+    return bigtableInstanceName.toTableNameStr(tableName.getNameAsString());
+  }
+
   // ****** TO BE IMPLEMENTED [end] ******
 
   /** {@inheritDoc} */
@@ -618,16 +671,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("abortProcedure"); // TODO
   }
 
-  /** {@inheritDoc} */
-  @Override
-  public CompletableFuture<Void> addColumnFamily(TableName tableName, ColumnFamilyDescriptor columnFamilyDesc) {
-    return modifyColumns(tableName, Modification
-        .newBuilder()
-        .setId(columnFamilyDesc.getNameAsString())
-        .setCreate(tableAdapter2x.toColumnFamily(columnFamilyDesc))
-        .build());
-  }
-
   /** {@inheritDoc} */
   @Override
   public CompletableFuture<Void> addReplicationPeer(String arg0, ReplicationPeerConfig arg1) {
@@ -1128,46 +1171,4 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   public CompletableFuture<Boolean> splitSwitch(boolean arg0) {
     throw new UnsupportedOperationException("splitSwitch"); // TODO
   }
-
-  private BigtableClusterName getSnapshotClusterName() throws IOException {
-    if (bigtableSnapshotClusterName == null) {
-      try {
-        bigtableSnapshotClusterName = getClusterName();
-      } catch (IllegalStateException e) {
-        throw new IllegalStateException(
-            "Failed to determine which cluster to use for snapshots, please configure it using "
-                + BigtableOptionsFactory.BIGTABLE_SNAPSHOT_CLUSTER_ID_KEY);
-      }
-    }
-    return bigtableSnapshotClusterName;
-  }
-
-  /**
-   * <p>modifyColumns.</p>
-   *
-   * @param tableName a {@link org.apache.hadoop.hbase.TableName} object.
-   * @param modification an array of {@link com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest.Modification} objectss
-   */
-  private CompletableFuture<Void> modifyColumns(TableName tableName, Modification... modifications) {
-    ModifyColumnFamiliesRequest request = ModifyColumnFamiliesRequest
-        .newBuilder()
-        .addAllModifications(Arrays.asList(modifications))
-        .setName(toBigtableName(tableName))
-        .build();
-    return bigtableTableAdminClient.modifyColumnFamilyAsync(request).thenApply(r -> null);
-  }
-
-  private BigtableClusterName getClusterName() throws IOException {
-    return asyncConnection.getSession().getClusterName();
-  }
-
-  /**
-   * <p>toBigtableName.</p>
-   *
-   * @param tableName a {@link org.apache.hadoop.hbase.TableName} object.
-   * @return a {@link java.lang.String} object.
-   */
-  private String toBigtableName(TableName tableName) {
-    return bigtableInstanceName.toTableNameStr(tableName.getNameAsString());
-  }
 }


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
index 65d875084..0984baa20 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
@@ -28,9 +28,21 @@ public final class ByteStringComparator implements Comparator<ByteString>{
 
   @Override
   public int compare(ByteString key1, ByteString key2) {
-    int size1 = key1.size();
-    int size2 = key2.size();
-    int size = Math.min(size1, size2);
+    if (key1 == null) {
+      if (key2 == null) {
+        return 0;
+      } else {
+        return 1;
+      }
+    } else if (key2 == null) {
+      return -1;
+    }
+
+    if (key1 == key2) {
+      return 0;
+    }
+
+    int size = Math.min(key1.size(), key2.size());
 
     for (int i = 0; i < size; i++) {
       // compare bytes as unsigned
@@ -42,6 +54,6 @@ public final class ByteStringComparator implements Comparator<ByteString>{
         return comparison;
       }
     }
-    return Integer.compare(size1, size2);
+    return Integer.compare(key1.size(), key2.size());
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
index b52faddd1..c28506069 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
@@ -36,12 +36,6 @@ public class ReadRowsRequestManagerTest {
 
   static final ByteString BLANK = ByteString.EMPTY;
 
-  static FlatRow buildRow(String rowKey) {
-    return FlatRow.newBuilder()
-        .withRowKey(ByteString.copyFromUtf8(rowKey))
-        .build();
-  }
-
   private static ReadRowsRequest createRequest(RowRange range) {
     return ReadRowsRequest.newBuilder().setRows(RowSet.newBuilder().addRowRanges(range)).build();
   }
@@ -62,14 +56,13 @@ public class ReadRowsRequestManagerTest {
     return RowSet.newBuilder().addAllRowKeys(keys).build();
   }
 
-
   /**
-   * Test a single, full table scan scenario for {@Link ResumingStreamingResultScanner#filterRows()}
+   * Test a single, full table scan scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}
    * .
    * @throws IOException
    */
   @Test
-  public void test_filterRows_testAllRange() throws IOException{
+  public void test_filterRows_testAllRange() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
 
     ReadRowsRequest originalRequest =
@@ -83,11 +76,33 @@ public class ReadRowsRequestManagerTest {
   }
 
   /**
-   * Test rowKeys scenario for {@Link ReadRowsRequestManager#getUpdatedRequest()}.
+   * Test a single, full table scan scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}}
+   * .
+   */
+   @Test
+  public void test_filterRows_empty() {
+     ByteString key1 = ByteString.copyFrom("row1".getBytes());
+
+     ReadRowsRequest originalRequest = createRequest(RowRange.getDefaultInstance());
+
+     ReadRowsRequestManager underTest = new ReadRowsRequestManager(originalRequest);
+
+     underTest.updateLastFoundKey(key1);
+     RowSet actual = underTest.buildUpdatedRequest().getRows();
+     Assert.assertEquals(0, actual.getRowKeysCount());
+     Assert.assertEquals(1, actual.getRowRangesCount());
+     RowRange actualRange = actual.getRowRanges(0);
+     Assert.assertEquals(key1, actualRange.getStartKeyOpen());
+     RowRange.EndKeyCase endKeyCase = actualRange.getEndKeyCase();
+     Assert.assertTrue(endKeyCase == RowRange.EndKeyCase.ENDKEY_NOT_SET);
+   }
+
+  /**
+   * Test rowKeys scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}}.
    * @throws IOException
    */
   @Test
-  public void test_filterRows_rowKeys() throws IOException{
+  public void test_filterRows_rowKeys() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
     ByteString key2 = ByteString.copyFrom("row2".getBytes());
     ByteString key3 = ByteString.copyFrom("row3".getBytes());
@@ -104,11 +119,11 @@ public class ReadRowsRequestManagerTest {
   }
 
   /**
-   * Test multiple rowset filter scenarios for {@Link ReadRowsRequestManager#getUpdatedRequest()}.
+   * Test multiple rowset filter scenarios for {@link ReadRowsRequestManager#buildUpdatedRequest()}}.
    * @throws IOException
    */
   @Test
-  public void test_filterRows_multiRowSetFilters() throws IOException{
+  public void test_filterRows_multiRowSetFilters() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
     ByteString key2 = ByteString.copyFrom("row2".getBytes());
     ByteString key3 = ByteString.copyFrom("row3".getBytes());
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
index 8bee5ffe6..9be8bc649 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.grpc.scanner;
 
 import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -23,7 +24,6 @@ import java.util.Arrays;
 
 import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
-import com.google.cloud.bigtable.grpc.scanner.RowMerger;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.BytesValue;
 import com.google.protobuf.StringValue;
@@ -49,7 +49,7 @@ public class RowMergerTest {
   }
 
   @Test
-  public void testOneCellRow() {
+  public void testOneCellRow_success() {
     CellChunk cellChunk1 = createCell("row_key1", "family", "qualifier", "value", 1, true);
     CellChunk cellChunk2 = createCell("row_key2", "family", "qualifier", "value", 1, true);
     RowMerger underTest = new RowMerger(observer);
@@ -57,6 +57,17 @@ public class RowMergerTest {
     underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk2).build());
     verify(observer, times(1)).onNext(eq(toRow(cellChunk1)));
     verify(observer, times(1)).onNext(eq(toRow(cellChunk2)));
+    verify(observer, times(0)).onError(any(Exception.class));
+  }
+
+  @Test
+  public void testOneCellRow_reverse_key() {
+    CellChunk cellChunk1 = createCell("row_key2", "family", "qualifier", "value", 1, true);
+    CellChunk cellChunk2 = createCell("row_key1", "family", "qualifier", "value", 1, true);
+    RowMerger underTest = new RowMerger(observer);
+    underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk1).build());
+    underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk2).build());
+    verify(observer, times(1)).onError(any(IllegalArgumentException.class));
   }
 
   @Test
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
index fdaf71041..99eb69bed 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
@@ -56,7 +56,8 @@ public class RowMergerPerf {
     final int qualifiersPerFamily = 15;
     Preconditions.checkArgument(size > 0, "size has to be > 0.");
     final ByteString rowKey = ByteString.copyFrom(Bytes.toBytes("rowkey-0"));
-    for (int i = 0; i < cellCount; i++) {
+    int quals = Math.max(1, cellCount - 1);
+    for (int i = 0; i < quals; i++) {
       CellChunk.Builder contentChunk =
           CellChunk.newBuilder()
               .setRowKey(i == 0 ? rowKey : ByteString.EMPTY)
@@ -64,13 +65,23 @@ public class RowMergerPerf {
                   .setValue(ByteString.copyFromUtf8("Qualifier" + (i % qualifiersPerFamily))))
               .setValue(ByteString.copyFrom(RandomStringUtils.randomAlphanumeric(size).getBytes()))
               .setTimestampMicros(330020L)
-              .setCommitRow(i == cellCount - 1);
+              .setCommitRow(i == quals - 1);
       if (i % qualifiersPerFamily == 0) {
         contentChunk.setFamilyName(StringValue.newBuilder().setValue("Family" + (i / qualifiersPerFamily)));
       }
 
       readRowsResponse.addChunks(contentChunk);
     }
+
+    // This adds a new row to test performance of comparing two different row keys.
+    readRowsResponse.addChunks(CellChunk.newBuilder()
+        .setRowKey(ByteString.copyFromUtf8("rowkey-1"))
+        .setQualifier(BytesValue.newBuilder()
+            .setValue(ByteString.copyFromUtf8("Qualifier0")))
+        .setFamilyName(StringValue.newBuilder().setValue("Family1").build())
+        .setValue(ByteString.copyFrom(RandomStringUtils.randomAlphanumeric(size).getBytes()))
+        .setTimestampMicros(330020L)
+        .setCommitRow(true));
     return readRowsResponse.build();
   }