fix: Add paging to hbase client - #4166

private static Result createRow(String key) {
    return Result.create(
        ImmutableList.<Cell>of(
            new com.google.cloud.bigtable.hbase.adapters.read.RowCell(
                Bytes.toBytes(key),
                Bytes.toBytes("cf"),
                Bytes.toBytes("q"),
                10L,
                Bytes.toBytes("value"),
                ImmutableList.of("label"))));
  }
  @Test
  public void testReadPaginatedRows() throws IOException {
    Query query = Query.create(TABLE_ID).range("a", "z");
    when(mockDataClient.readRowsCallable(Mockito.<RowResultAdapter>any()))
        .thenReturn(mockStreamingCallable);

    // First Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("a"));
      observer.onResponse(createRow("b"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(eq(Query.create(TABLE_ID).range("a", "z").limit(2)), any(), any());

    // 2nd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("c"));
      observer.onResponse(createRow("d"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
            eq(
              Query.create(TABLE_ID)
                .range(ByteStringRange.unbounded().startOpen("b").endOpen("z")).limit(2)),
            any(), any());

    // 3rd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("e"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
        eq(
          Query.create(TABLE_ID)
            .range(ByteStringRange.unbounded().startOpen("d").endOpen("z")).limit(2)), any(), any());

    // 3rd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
            eq(
                Query.create(TABLE_ID)
                    .range(ByteStringRange.unbounded().startOpen("e").endOpen("z")).limit(2)), any(), any());

    ResultScanner resultScanner = dataClientWrapper.readRows(query.createPaginator(2), 1000);

    assertThat(resultScanner)
        .comparingElementsUsing(Correspondence.transforming((Result r) -> new String(r.getRow()), "row key"))
        .containsExactly("a", "b", "c", "d", "e");
  }

  public void testReadRows_Errors() throws IOException {
    Query query = Query.create(TABLE_ID).rowKey(ROW_KEY);
    when(mockDataClient.readRowsCallable(any(RowResultAdapter.class)))
        .thenReturn(mockStreamingCallable);
    when(mockStreamingCallable.call(any(Query.class), any(GrpcCallContext.class)))
        .thenReturn(serverStream);
    when(serverStream.iterator())
        .thenReturn(new Iterator<Result>() {
          @Override
          public boolean hasNext() {
            return true;
          }
          @Override
          public Result next() {
            throw new InternalException("fake error", null, GrpcStatusCode.of(Code.INTERNAL), false);
          }
        })
        .thenReturn(ImmutableList.<Result>of().iterator());

    assertThrows(Exception.class, () -> dataClientWrapper.readRows(query).next());

    ResultScanner noRowsResultScanner = dataClientWrapper.readRows(query);
    assertNull(noRowsResultScanner.next());
    noRowsResultScanner.close();

    verify(mockDataClient, times(2)).readRowsCallable(Mockito.<RowResultAdapter>any());
    verify(serverStream, times(2)).iterator();
    verify(mockStreamingCallable, times(2))
        .call(any(Query.class), any(GrpcCallContext.class));
  }

 public void testReadRowsLowMemory() throws IOException {
    Query query = Query.create(TABLE_ID);
    when(mockDataClient.readRowsCallable(any(RowResultAdapter.class)))
        .thenReturn(mockStreamingCallable);

    StreamController mockController = Mockito.mock(StreamController.class);
    doAnswer(invocation -> {
          cancelled.set(true);
          return null;
        })
        .when(mockController)
        .cancel();


    // Generate
    doAnswer(
        (Answer<Void>) invocation -> {
          ResponseObserver<Result> observer = invocation.getArgument(1);
          observer.onStart(mockController);

          for(int i=0; i < 1000 && !cancelled.get(); i++) {
            observer.onResponse(createRow(String.format("row%010d", i)));
            Thread.sleep(10);
          }
          observer.onComplete();
          return null;
        })
        .doAnswer(
        (Answer<Void>) invocation -> {
          ResponseObserver<Result> observer = invocation.getArgument(1);
          observer.onComplete();
          return null;
        })
        .when(mockStreamingCallable)
        .call(any(), any(), any());

    ResultScanner resultScanner = dataClientWrapper.readRows(query.createPaginator(100), 3);
    // Consume the stream
    Lists.newArrayList(resultScanner);

    verify(mockStreamingCallable, times(2))
        .call(
            any(Query.class),
            any(ResponseObserver.class),
            any(GrpcCallContext.class));
    assertTrue(cancelled.get());
  }