@Test
public void testReadRowAsync() throws Exception {

  ServerStreamingCallable<Query, Row> serverStreaming = mock(ServerStreamingCallable.class);
  UnaryCallable<Query, List<Row>> unaryCallable = mock(UnaryCallable.class);
  List<Row> expectedRows =
      ImmutableList.of(
          Row.create(
              ByteString.copyFromUtf8(ROW_KEY),
              ImmutableList.of(
                  RowCell.create(COL_FAMILY, QUALIFIER_1, TIMESTAMP, LABELS, VALUE_1))),
          Row.create(
              ByteString.copyFromUtf8("row-key-2"),
              ImmutableList.of(
                  RowCell.create(COL_FAMILY, QUALIFIER_2, TIMESTAMP, LABELS, VALUE_2))));
  when(dataClientV2.readRowsCallable()).thenReturn(serverStreaming);
  when(serverStreaming.all()).thenReturn(unaryCallable);
  when(unaryCallable.futureCall(request)).thenReturn(immediateFuture(expectedRows));
  ApiFuture<List<Row>> actualRows = dataGCJClient.readRowsAsync(request);
  assertEquals(expectedRows, actualRows.get());
  verify(dataClientV2).readRowsCallable();
  verify(serverStreaming).all();
  verify(unaryCallable).futureCall(request);
}