@Test
public void testReadModifyWriteAsyncWithOneRow() throws ExecutionException, InterruptedException {

  String family_1 = "col-family-1";
  String family_2 = "col-family-2";
  String family_3 = "col-family-3";
  ByteString qualifier_1 = ByteString.copyFromUtf8("test-qualifier_1-1");
  ByteString qualifier_2 = ByteString.copyFromUtf8("test-qualifier_1-2");
  ByteString qualifier_3 = ByteString.copyFromUtf8("test-qualifier_1-2");
  ByteString value_1 = ByteString.copyFromUtf8("test-values-1");
  ByteString value_2 = ByteString.copyFromUtf8("test-values-2");
  ByteString value_3 = ByteString.copyFromUtf8("test-values-3");
  ByteString value_4 = ByteString.copyFromUtf8("test-values-4");
  ByteString value_5 = ByteString.copyFromUtf8("test-values-5");
  ByteString value_6 = ByteString.copyFromUtf8("test-values-6");
  Row row =
      Row.newBuilder()
          .setKey(ROW_KEY)
          .addFamilies(
              Family.newBuilder()
                  .setName(family_1)
                  .addColumns(
                      Column.newBuilder()
                          .setQualifier(qualifier_1)
                          // First Cell
                          .addCells(
                              Cell.newBuilder()
                                  .setTimestampMicros(11_111L)
                                  .setValue(value_1)
                                  .addLabels("label-1"))
                          // Same Cell with another timestamp and value
                          .addCells(
                              Cell.newBuilder()
                                  .setTimestampMicros(22_222L)
                                  .setValue(value_2)
                                  .addLabels("label-2")))
                  .addColumns(
                      Column.newBuilder()
                          .setQualifier(qualifier_2)
                          // With label
                          .addCells(
                              Cell.newBuilder()
                                  .setTimestampMicros(11_111L)
                                  .setValue(value_3)
                                  .addLabels("label-3")
                                  .addLabels("label-4"))
                          // Same family, same timestamp, but different column.
                          .addCells(
                              Cell.newBuilder().setTimestampMicros(22_222L).setValue(value_4)))
                  .build())
          .addFamilies(
              Family.newBuilder()
                  .setName(family_2)
                  .addColumns(
                      Column.newBuilder()
                          .setQualifier(qualifier_1)
                          // Same column, same timestamp, but different family.
                          .addCells(
                              Cell.newBuilder().setTimestampMicros(11_111L).setValue(value_5))))
          .addFamilies(
              Family.newBuilder()
                  .setName(family_3)
                  .addColumns(
                      Column.newBuilder()
                          .setQualifier(qualifier_3)
                          // Same timestamp, but different family and column.
                          .addCells(Cell.newBuilder().setValue(value_6).addLabels("label-6"))))
          .build();
  ReadModifyWriteRow readModify = ReadModifyWriteRow.create(TABLE_ID, "test-key");
  ReadModifyWriteRowRequest request = readModify.toProto(REQUEST_CONTEXT);
  ReadModifyWriteRowResponse response =
      ReadModifyWriteRowResponse.newBuilder().setRow(row).build();
  ListenableFuture<ReadModifyWriteRowResponse> listenableResponse =
      Futures.immediateFuture(response);
  when(delegate.readModifyWriteRowAsync(request)).thenReturn(listenableResponse);
  Result result = dataClientWrapper.readModifyWriteRowAsync(readModify).get();
  assertEquals(6, result.rawCells().length);
  List<org.apache.hadoop.hbase.Cell> cells1 =
      result.getColumnCells(family_1.getBytes(), qualifier_1.toByteArray());
  assertEquals(2, cells1.size());
  assertEquals(11L, cells1.get(0).getTimestamp());
  assertArrayEquals(value_1.toByteArray(), CellUtil.cloneValue(cells1.get(0)));
  assertEquals(Collections.singletonList("label-1"), ((RowCell) cells1.get(0)).getLabels());
  assertEquals(22L, cells1.get(1).getTimestamp());
  assertArrayEquals(value_2.toByteArray(), CellUtil.cloneValue(cells1.get(1)));
  assertEquals(Collections.singletonList("label-2"), ((RowCell) cells1.get(1)).getLabels());
  List<org.apache.hadoop.hbase.Cell> cells2 =
      result.getColumnCells(family_1.getBytes(), qualifier_2.toByteArray());
  assertEquals(2, cells2.size());
  assertEquals(11L, cells2.get(0).getTimestamp());
  assertArrayEquals(value_3.toByteArray(), CellUtil.cloneValue(cells2.get(0)));
  assertEquals(ImmutableList.of("label-3", "label-4"), ((RowCell) cells2.get(0)).getLabels());
  assertEquals(22L, cells2.get(1).getTimestamp());
  assertArrayEquals(value_4.toByteArray(), CellUtil.cloneValue(cells2.get(1)));
  List<org.apache.hadoop.hbase.Cell> cells3 =
      result.getColumnCells(family_2.getBytes(), qualifier_1.toByteArray());
  assertEquals(1, cells3.size());
  assertArrayEquals(value_5.toByteArray(), CellUtil.cloneValue(cells3.get(0)));
  List<org.apache.hadoop.hbase.Cell> cells4 =
      result.getColumnCells(family_3.getBytes(), qualifier_3.toByteArray());
  assertEquals(1, cells4.size());
  assertArrayEquals(value_6.toByteArray(), CellUtil.cloneValue(cells4.get(0)));
  assertEquals(Collections.singletonList("label-6"), ((RowCell) cells4.get(0)).getLabels());
}