Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #2035 (2018-12-27 15:59:28)

IBigtableDataClient remaining method Implementations
## What this PR contains

I am considering two steps for migrating to ApiFuture:
  **1st step** would be to adapt to GCJ `v2.models` for all operation than in 
   **2nd step** we can migrate to `ApiFuture`,

In this PR we have adapted to `ListenableFuture` in `IBigtableDataClient`.

 - Added implementation for these methods:
  1. mutateRowAsync
  2. readModifyWriteRow
  3. readModifyWriteRowAsync,
  4. checkAndMutateRow
  5. checkAndMutateRowAsync
- Added utility from `CheckAndMutationUtil#wasMutationApplied` in `BigtableDataClientWrappperImpl`, as `CheckAndMutationUtil` is in child module.
- Added `convert()` for conversion between `com.google.bigtable.v2.Row` to `v2.Models.Row`.


Patch:
diff --git a/README.md b/README.md
index b710d671c..e8ee38e19 100644
--- a/README.md
+++ b/README.md
@@ -122,7 +122,7 @@ maven repository. When you disable that profile, attach workspace sources to the
 
 | Name | Language | Repository | latest version | status |
 | --- | --- | --- | --- | --- |
-| Cloud Bigtable Examples | Java, others | [GoogleCloudPlatform/cloud-bitgable-examples][maven-examples-repo] | | |
+| Cloud Bigtable Examples | Java, others | [GoogleCloudPlatform/cloud-bigtable-examples][maven-examples-repo] | | |
 | HBase client | Java | [GoogleCloudPlatform/cloud-bigtable-client][maven-hbase-client-repo] | [![Maven][maven-hbase-shield]][maven-hbase-client-maven-search] | GA |
 | Cloud Bigtable GoLang | Go | [googleapis/google-cloud-go](https://github.com/googleapis/google-cloud-go) | N/A | GA |
 | Cloud Bigtable Java | java | [googleapis/google-cloud-java](http://github.com/googleapis/google-cloud-java) | [![Maven][maven-google-cloud-java-shield]][maven-google-cloud-java-maven-search] | Beta |
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
index d952330ca..0866ca7ee 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
@@ -15,12 +15,11 @@
  */
 package com.google.cloud.bigtable.core;
 
-import com.google.api.core.ApiFuture;
 import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
 import com.google.cloud.bigtable.data.v2.models.Row;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
-
+import com.google.common.util.concurrent.ListenableFuture;
 import java.util.concurrent.ExecutionException;
 
 /**
@@ -41,17 +40,18 @@ public interface IBigtableDataClient {
    * Mutate a row atomically.
    *
    * @param rowMutation a {@link RowMutation} model object.
-   * @return a {@link ApiFuture} of type {@link Void} will be set when request is
+   * @return a {@link ListenableFuture} of type {@link Void} will be set when request is
    *     successful otherwise exception will be thrown.
    * @throws InterruptedException if any.
    */
-  ApiFuture<Void> mutateRowAsync(RowMutation rowMutation) throws InterruptedException;
+  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
+  ListenableFuture<Void> mutateRowAsync(RowMutation rowMutation) throws InterruptedException;
 
   /**
    * Perform an atomic read-modify-write operation on a row.
    *
    * @param readModifyWriteRow a {@link ReadModifyWriteRow} model object.
-   * @return Row a modified row.
+   * @return {@link Row} a modified row.
    * @throws ExecutionException if any.
    * @throws InterruptedException if any.
    */
@@ -62,11 +62,12 @@ public interface IBigtableDataClient {
    * Perform an atomic read-modify-write operation on a row.
    *
    * @param readModifyWriteRow a {@link ReadModifyWriteRow} model object.
-   * @return a {@link ApiFuture} of type {@link Row} will be set when request is
+   * @return a {@link ListenableFuture} of type {@link Row} will be set when request is
    *     successful otherwise exception will be thrown.
    * @throws InterruptedException if any.
    */
-  ApiFuture<Row> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) throws InterruptedException;
+  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
+  ListenableFuture<Row> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) throws InterruptedException;
 
   /**
    * Creates BulMutation batcher.
@@ -77,16 +78,17 @@ public interface IBigtableDataClient {
    * Mutate a row atomically dependent on a precondition.
    *
    * @param conditionalRowMutation a {@link ConditionalRowMutation} model object.
-   * @return a {@link ApiFuture} of type {@link Boolean} will be set when request is
+   * @return a {@link ListenableFuture} of type {@link Boolean} will be set when request is
    *     successful otherwise exception will be thrown.
    */
-  ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation);
+  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
+  ListenableFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation);
 
   /**
    * Mutate a row atomically dependent on a precondition.
    *
    * @param conditionalRowMutation a {@link ConditionalRowMutation} model object.
-   * @return Boolean returns true if predicate returns any result.
+   * @return returns true if predicate returns any result.
    * @throws ExecutionException if any.
    * @throws InterruptedException if any.
    */
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
index ddf1314ef..bab3a03d3 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
@@ -15,8 +15,14 @@
  */
 package com.google.cloud.bigtable.grpc;
 
-import com.google.api.core.ApiFuture;
+import com.google.bigtable.v2.Cell;
+import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.CheckAndMutateRowResponse;
+import com.google.bigtable.v2.Column;
+import com.google.bigtable.v2.Family;
 import com.google.bigtable.v2.MutateRowRequest;
+import com.google.bigtable.v2.MutateRowResponse;
+import com.google.bigtable.v2.ReadModifyWriteRowResponse;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.core.IBigtableDataClient;
 import com.google.cloud.bigtable.core.IBulkMutation;
@@ -25,11 +31,18 @@ import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.InstanceName;
 import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
 import com.google.cloud.bigtable.data.v2.models.Row;
+import com.google.cloud.bigtable.data.v2.models.RowCell;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import javax.annotation.Nullable;
 
 /**
  * This class implements the {@link IBigtableDataClient} interface which provides access to google cloud
- * java
+ * java.
  */
 public class BigtableDataClientWrapper implements IBigtableDataClient {
 
@@ -46,39 +59,100 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
         );
   }
 
+  /** {@inheritDoc} */
   @Override
   public void mutateRow(RowMutation rowMutation) {
     MutateRowRequest mutateRowRequest = rowMutation.toProto(requestContext);
     delegate.mutateRow(mutateRowRequest);
   }
 
+  /** {@inheritDoc} */
   @Override
-  public ApiFuture<Void> mutateRowAsync(RowMutation rowMutation) {
-    throw new UnsupportedOperationException("Not implemented yet");
+  public ListenableFuture<Void> mutateRowAsync(RowMutation rowMutation) {
+    ListenableFuture<MutateRowResponse> response =
+        delegate.mutateRowAsync(rowMutation.toProto(requestContext));
+    return  Futures.transform(response, new Function<MutateRowResponse, Void>() {
+      @Nullable
+      @Override
+      public Void apply(@Nullable MutateRowResponse mutateRowResponse) {
+        return null;
+      }
+    }, MoreExecutors.directExecutor());
   }
 
+  /** {@inheritDoc} */
   @Override
   public Row readModifyWriteRow(ReadModifyWriteRow readModifyWriteRow) {
-    throw new UnsupportedOperationException("Not implemented yet");
+    ReadModifyWriteRowResponse response =
+        delegate.readModifyWriteRow(readModifyWriteRow.toProto(requestContext));
+    return transformResponse(response);
   }
 
+  /** {@inheritDoc} */
   @Override
-  public ApiFuture<Row> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) {
-    throw new UnsupportedOperationException("Not implemented yet");
+  public ListenableFuture<Row> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) {
+    ListenableFuture<ReadModifyWriteRowResponse> response =
+        delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext));
+    return Futures.transform(response, new Function<ReadModifyWriteRowResponse, Row>() {
+      @Override
+      public Row apply(ReadModifyWriteRowResponse readModifyRowResponse) {
+        return transformResponse(readModifyRowResponse);
+      }
+    }, MoreExecutors.directExecutor());
   }
 
+  /** {@inheritDoc} */
   @Override
   public IBulkMutation createBulkMutationBatcher() {
     throw new UnsupportedOperationException("Not implemented yet");
   }
 
+  /** {@inheritDoc} */
   @Override
-  public ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation) {
-    throw new UnsupportedOperationException("Not implemented yet");
+  public ListenableFuture<Boolean> checkAndMutateRowAsync(
+      ConditionalRowMutation conditionalRowMutation) {
+    final CheckAndMutateRowRequest request = conditionalRowMutation.toProto(requestContext);
+    final ListenableFuture<CheckAndMutateRowResponse> response =
+        delegate.checkAndMutateRowAsync(request);
+    return Futures.transform(response, new Function<CheckAndMutateRowResponse, Boolean>() {
+
+      @Override
+      public Boolean apply(CheckAndMutateRowResponse checkAndMutateRowResponse) {
+        return checkAndMutateRowResponse.getPredicateMatched();
+      }
+    }, MoreExecutors.directExecutor());
   }
 
+  /** {@inheritDoc} */
   @Override
   public Boolean checkAndMutateRow(ConditionalRowMutation conditionalRowMutation) {
-    throw new UnsupportedOperationException("Not implemented yet");
+    CheckAndMutateRowResponse response =
+        delegate.checkAndMutateRow(conditionalRowMutation.toProto(requestContext));
+    return response.getPredicateMatched();
+  }
+
+  /**
+   * This method converts instances of {@link ReadModifyWriteRowResponse} to {@link Row}.
+   *
+   * @param response an instance of {@link ReadModifyWriteRowResponse} type.
+   * @return an instance of {@link Row}.
+   */
+  private Row transformResponse(ReadModifyWriteRowResponse response) {
+    ImmutableList.Builder<RowCell> rowCells  = ImmutableList.builder();
+
+    for (Family family : response.getRow().getFamiliesList()) {
+      for (Column column : family.getColumnsList()) {
+        for (Cell cell : column.getCellsList()) {
+          rowCells.add(
+              RowCell.create(
+                  family.getName(),
+                  column.getQualifier(),
+                  cell.getTimestampMicros(),
+                  cell.getLabelsList(),
+                  cell.getValue()));
+        }
+      }
+    }
+    return Row.create(response.getRow().getKey(), rowCells.build());
   }
 }
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
index 8463767e0..4e655566a 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
@@ -62,7 +62,7 @@ import org.apache.hadoop.io.serializer.WritableSerialization;
  *    --zone=[ZONE] \
  *    --bigtableInstanceId=[INSTANCE] \
  *    --bigtableTableId=[TABLE] \
- *    --destination=gs://[BUCLET]/[EXPORT_PATH] \
+ *    --destination=gs://[BUCKET]/[EXPORT_PATH] \
  *    --maxNumWorkers=[nodes * 10]"
  * }
  * </pre>


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
index f6690efc2..9e5aea397 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
@@ -15,19 +15,40 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.bigtable.v2.Cell;
+import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.CheckAndMutateRowResponse;
+import com.google.bigtable.v2.Column;
+import com.google.bigtable.v2.Family;
 import com.google.bigtable.v2.MutateRowRequest;
 import com.google.bigtable.v2.MutateRowResponse;
+import com.google.bigtable.v2.ReadModifyWriteRowRequest;
+import com.google.bigtable.v2.ReadModifyWriteRowResponse;
+import com.google.bigtable.v2.Row;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.data.v2.internal.RequestContext;
+import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.InstanceName;
+import com.google.cloud.bigtable.data.v2.models.Mutation;
+import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
+import com.google.cloud.bigtable.data.v2.models.RowCell;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import java.util.Collections;
+import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.Mockito;
+import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -37,27 +58,225 @@ public class TestBigtableDataClientWrapper {
   private static final String INSTANCE_ID = "instanceId";
   private static final String TABLE_ID = "tableId";
   private static final String APP_PROFILE_ID = "appProfileId";
+  private static final ByteString QUALIFIER_1 = ByteString.copyFromUtf8("qualifier1");
+  private static final ByteString QUALIFIER_2 = ByteString.copyFromUtf8("qualifier2");
+  private static final int TIMESTAMP = 12345;
+  private static final String LABEL = "label";
+  private static final List<String> LABEL_LIST = Collections.singletonList(LABEL);
+  private static final ByteString VALUE = ByteString.copyFromUtf8("test-value");
+  private static final ByteString ROW_KEY = ByteString.copyFromUtf8("test-key");
+
   private static final RequestContext REQUEST_CONTEXT =
       RequestContext.create(InstanceName.of(PROJECT_ID, INSTANCE_ID), APP_PROFILE_ID);
-  private BigtableDataClient bigtableDataClient;
-  private BigtableOptions options;
-  private BigtableDataClientWrapper bigtableDataClientWrapper;
+
+  private BigtableOptions options =
+      BigtableOptions.builder().setProjectId(PROJECT_ID).setInstanceId(INSTANCE_ID)
+        .setAppProfileId(APP_PROFILE_ID).build();
+  @Mock
+  private BigtableDataClient client;
+
+  private BigtableDataClientWrapper clientWrapper;
 
   @Before
   public void setUp() {
-    bigtableDataClient = Mockito.mock(BigtableDataClient.class);
-    options = BigtableOptions.builder().setProjectId(PROJECT_ID).setInstanceId(INSTANCE_ID)
-        .setAppProfileId(APP_PROFILE_ID).build();
-    bigtableDataClientWrapper = new BigtableDataClientWrapper(bigtableDataClient, options);
+    clientWrapper = new BigtableDataClientWrapper(client, options);
   }
 
   @Test
   public void testMutateRow() {
     RowMutation rowMutation = RowMutation.create(TABLE_ID, "key");
     MutateRowRequest mutateRowRequest = rowMutation.toProto(REQUEST_CONTEXT);
-    when(bigtableDataClient.mutateRow(mutateRowRequest))
-        .thenReturn(MutateRowResponse.getDefaultInstance());
-    bigtableDataClientWrapper.mutateRow(rowMutation);
-    verify(bigtableDataClient).mutateRow(mutateRowRequest);
+    when(client.mutateRow(mutateRowRequest)).thenReturn(MutateRowResponse.getDefaultInstance());
+    clientWrapper.mutateRow(rowMutation);
+    verify(client).mutateRow(mutateRowRequest);
+  }
+
+  @Test
+  public void testMutateRowAsync() {
+    RowMutation rowMutation = RowMutation.create(TABLE_ID, "key");
+    MutateRowRequest request = rowMutation.toProto(REQUEST_CONTEXT);
+    ListenableFuture<MutateRowResponse> response =
+        Futures.immediateFuture(MutateRowResponse.getDefaultInstance());
+    when(client.mutateRowAsync(request)).thenReturn(response);
+    clientWrapper.mutateRowAsync(rowMutation);
+    verify(client).mutateRowAsync(request);
+  }
+
+  @Test
+  public void testCheckMutateRow(){
+    Mutation mutation = Mutation.create();
+    mutation.setCell("family", "qualifier", "some other value");
+    ConditionalRowMutation conditonalMutation =
+        ConditionalRowMutation.create(TABLE_ID, "first" + "-row" + "-key").then(mutation);
+    CheckAndMutateRowRequest request = conditonalMutation.toProto(REQUEST_CONTEXT);
+    CheckAndMutateRowResponse response =
+        CheckAndMutateRowResponse.newBuilder().setPredicateMatched(true).build();
+    when(client.checkAndMutateRow(request)).thenReturn(response);
+    Boolean actual = clientWrapper.checkAndMutateRow(conditonalMutation);
+    verify(client).checkAndMutateRow(request);
+    assertTrue(actual);
+  }
+
+  @Test
+  public void testCheckMutateRowWhenNoPredicateMatch(){
+    Mutation mutation = Mutation.create();
+    mutation.setCell("family", "qualifier", "some other value");
+    ConditionalRowMutation conditonalMutation =
+        ConditionalRowMutation.create(TABLE_ID, "first" + "-row" + "-key").then(mutation);
+    CheckAndMutateRowRequest request = conditonalMutation.toProto(REQUEST_CONTEXT);
+    CheckAndMutateRowResponse response =
+        CheckAndMutateRowResponse.newBuilder().setPredicateMatched(false).build();
+    when(client.checkAndMutateRow(request)).thenReturn(response);
+    Boolean actual = clientWrapper.checkAndMutateRow(conditonalMutation);
+    verify(client).checkAndMutateRow(request);
+    assertFalse(actual);
+  }
+
+  @Test
+  public void testCheckMutateRowAsync() throws Exception{
+    Mutation mutation = Mutation.create();
+    mutation.setCell("family", "qualifier", "some other value");
+    ConditionalRowMutation conditonalMutation =
+        ConditionalRowMutation.create(TABLE_ID, "first" + "-row" + "-key").then(mutation);
+    CheckAndMutateRowRequest request = conditonalMutation.toProto(REQUEST_CONTEXT);
+    CheckAndMutateRowResponse response =
+        CheckAndMutateRowResponse.newBuilder().setPredicateMatched(true).build();
+    ListenableFuture<CheckAndMutateRowResponse> future = Futures.immediateFuture(response);
+
+    when(client.checkAndMutateRowAsync(request)).thenReturn(future);
+    ListenableFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
+    verify(client).checkAndMutateRowAsync(request);
+    assertTrue(actual.get());
+  }
+
+  @Test
+  public void testCheckMutateRowAsyncWhenNoPredicateMatch() throws Exception{
+    Mutation mutation = Mutation.create();
+    mutation.setCell("family", "qualifier", "some other value");
+    ConditionalRowMutation conditonalMutation =
+        ConditionalRowMutation.create(TABLE_ID, "first" + "-row" + "-key").then(mutation);
+    CheckAndMutateRowRequest request = conditonalMutation.toProto(REQUEST_CONTEXT);
+    CheckAndMutateRowResponse response =
+        CheckAndMutateRowResponse.newBuilder().setPredicateMatched(false).build();
+    ListenableFuture<CheckAndMutateRowResponse> future = Futures.immediateFuture(response);
+
+    when(client.checkAndMutateRowAsync(request)).thenReturn(future);
+    ListenableFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
+    verify(client).checkAndMutateRowAsync(request);
+    assertFalse(actual.get());
+  }
+
+  @Test
+  public void testReadModifyWriteWithOneCell(){
+    ReadModifyWriteRow readModify = ReadModifyWriteRow.create(TABLE_ID, "test-key");
+    ReadModifyWriteRowRequest request = readModify.toProto(REQUEST_CONTEXT);
+    Row expectedRow = buildRow();
+    ReadModifyWriteRowResponse response =
+        ReadModifyWriteRowResponse.newBuilder().setRow(expectedRow).build();
+    RowCell rowCell = RowCell.create("firstFamily", QUALIFIER_1, TIMESTAMP, LABEL_LIST, VALUE);
+    com.google.cloud.bigtable.data.v2.models.Row modelRow =
+        com.google.cloud.bigtable.data.v2.models.Row.create(ROW_KEY, Collections.singletonList(rowCell));
+
+    when(client.readModifyWriteRow(request)).thenReturn(response);
+    com.google.cloud.bigtable.data.v2.models.Row actualRow =
+        clientWrapper.readModifyWriteRow(readModify);
+    assertEquals(modelRow, actualRow);
+    verify(client).readModifyWriteRow(request);
+  }
+
+  @Test
+  public void testReadModifyWriteAsync() throws Exception{
+    ReadModifyWriteRow readModify = ReadModifyWriteRow.create(TABLE_ID, "test-key");
+    ReadModifyWriteRowRequest request = readModify.toProto(REQUEST_CONTEXT);
+    Row expectedRow = buildRow();
+    ReadModifyWriteRowResponse response =
+        ReadModifyWriteRowResponse.newBuilder().setRow(expectedRow).build();
+    ListenableFuture<ReadModifyWriteRowResponse> listenableResponse =
+        Futures.immediateFuture(response);
+    RowCell rowCell = RowCell.create("firstFamily", QUALIFIER_1, TIMESTAMP, LABEL_LIST, VALUE);
+    com.google.cloud.bigtable.data.v2.models.Row modelRow =
+        com.google.cloud.bigtable.data.v2.models.Row.create(ROW_KEY, Collections.singletonList(rowCell));
+
+    when(client.readModifyWriteRowAsync(request)).thenReturn(listenableResponse);
+    ListenableFuture<com.google.cloud.bigtable.data.v2.models.Row> output =
+        clientWrapper.readModifyWriteRowAsync(readModify);
+    assertEquals(modelRow, output.get());
+    verify(client).readModifyWriteRowAsync(request);
+  }
+
+  private static Row buildRow() {
+    Cell cell = Cell.newBuilder()
+        .setValue(VALUE)
+        .setTimestampMicros(TIMESTAMP)
+        .addLabels(LABEL)
+        .build();
+    return Row.newBuilder()
+        .setKey(ROW_KEY)
+        .addFamilies(Family.newBuilder()
+            .setName("firstFamily")
+            .addColumns(Column.newBuilder()
+                .setQualifier(QUALIFIER_1)
+                .addCells(cell)
+                .build())
+            .build())
+        .build();
+  }
+
+  @Test
+  public void testReadModifyWriteWithMultipleCell(){
+    Row row = Row.newBuilder()
+        .setKey(ROW_KEY)
+        .addFamilies(Family.newBuilder()
+            .setName("firstFamily")
+            .addColumns(Column.newBuilder()
+                .setQualifier(QUALIFIER_1)
+                .addCells(Cell.newBuilder()
+                    .setValue(VALUE)
+                    .setTimestampMicros(TIMESTAMP)
+                    .addLabels(LABEL)
+                    .build())
+                .build())
+            .addColumns(Column.newBuilder()
+                .setQualifier(QUALIFIER_2)
+                .addCells(Cell.newBuilder()
+                    .setValue(VALUE)
+                    .setTimestampMicros(TIMESTAMP)
+                    .addLabels(LABEL)
+                    .build())
+                .addCells(Cell.newBuilder()
+                    .setValue(VALUE)
+                    .setTimestampMicros(54321)
+                    .addLabels(LABEL)
+                    .build())
+                .build())
+            .build())
+        .addFamilies(Family.newBuilder()
+            .setName("secondFamily")
+            .addColumns(Column.newBuilder()
+                .setQualifier(QUALIFIER_1)
+                .addCells(Cell.newBuilder()
+                    .setValue(VALUE)
+                    .setTimestampMicros(TIMESTAMP)
+                    .addLabels(LABEL)
+                    .build())
+                .build()))
+        .build();
+    ReadModifyWriteRowResponse response =
+        ReadModifyWriteRowResponse.newBuilder().setRow(row).build();
+    ReadModifyWriteRow readModify = ReadModifyWriteRow.create(TABLE_ID, "test-key");
+    ReadModifyWriteRowRequest request = readModify.toProto(REQUEST_CONTEXT);
+    ImmutableList.Builder<RowCell> rowCells = ImmutableList.builder();
+    rowCells.add(RowCell.create("firstFamily", QUALIFIER_1, TIMESTAMP, LABEL_LIST, VALUE));
+    rowCells.add(RowCell.create("firstFamily", QUALIFIER_2, TIMESTAMP, LABEL_LIST, VALUE));
+    rowCells.add(RowCell.create("firstFamily", QUALIFIER_2, 54321, LABEL_LIST, VALUE));
+    rowCells.add(RowCell.create("secondFamily", QUALIFIER_1, TIMESTAMP, LABEL_LIST, VALUE));
+    com.google.cloud.bigtable.data.v2.models.Row expectedModelRow =
+        com.google.cloud.bigtable.data.v2.models.Row.create(ROW_KEY, rowCells.build());
+
+    when(client.readModifyWriteRow(request)).thenReturn(response);
+    com.google.cloud.bigtable.data.v2.models.Row actualRow =
+        clientWrapper.readModifyWriteRow(readModify);
+    assertEquals(expectedModelRow, actualRow);
+    verify(client).readModifyWriteRow(request);
   }
 }