@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) {
          assertResult(EXPECTED_RESULT, result);
        }
        @Override
        public void onError(Throwable throwable) {
          assertEquals(readException, throwable);
        }
        @Override
        public void onCompleted() {}
      };
  when(mockDataClient.readRowsCallable(Mockito.<RowResultAdapter>any()))
      .thenReturn(mockStreamingCallable);
  doAnswer(
          new Answer() {
            int count = 0;
            @Override
            public Object answer(InvocationOnMock invocationOnMock) {
              ResponseObserver<Result> resObserver = invocationOnMock.getArgument(1);
              resObserver.onStart(null);
              resObserver.onResponse(EXPECTED_RESULT);
              if (count == 0) {
                resObserver.onComplete();
              } else {
                resObserver.onError(readException);
              }
              count++;
              return null;
            }
          })
      .when(mockStreamingCallable)
      .call(Mockito.<Query>any(), Mockito.<ResponseObserver<Result>>any());
  dataClientWrapper.readRowsAsync(request, resultStreamOb);
  dataClientWrapper.readRowsAsync(request, resultStreamOb);
  verify(mockDataClient, times(2)).readRowsCallable(Mockito.<RowResultAdapter>any());
  verify(mockStreamingCallable, times(2))
      .call(Mockito.<Query>any(), Mockito.<ResponseObserver<Result>>any());
}