@Test
public void testReadRowsAsyncWithStreamOb() {

  final Exception readException = new Exception();
  Query request = Query.create(TABLE_ID).rowKey(ROW_KEY);
  StreamObserver<Result> resultStreamOb =
      new StreamObserver<Result>() {
        @Override
        public void onNext(Result result) {
          assertArrayEquals(
              FLAT_ROW_ADAPTER.adaptResponse(SAMPLE_FLAT_ROW).rawCells(), result.rawCells());
        }
        @Override
        public void onError(Throwable throwable) {
          assertEquals(readException, throwable);
        }
        @Override
        public void onCompleted() {}
      };
  when(delegate.readFlatRows(
          Mockito.<ReadRowsRequest>any(), Mockito.<StreamObserver<FlatRow>>any()))
      .thenAnswer(
          new Answer<ScanHandler>() {
            @Override
            public ScanHandler answer(InvocationOnMock invocationOnMock) {
              StreamObserver<FlatRow> sb = invocationOnMock.getArgument(1);
              sb.onNext(SAMPLE_FLAT_ROW);
              sb.onError(readException);
              sb.onCompleted();
              return new ScanHandler() {
                @Override
                public void cancel() {}
              };
            }
          });
  dataClientWrapper.readRowsAsync(request, resultStreamOb);
  verify(delegate)
      .readFlatRows(Mockito.<ReadRowsRequest>any(), Mockito.<StreamObserver<FlatRow>>any());
}