RetryingReadRowsOperation extends AbstractRetryingOperation<ReadRowsRequest, ReadRowsResponse, String> implements ScanHandler { @Override public void onMessage(ReadRowsResponse message) { try { resetStatusBasedBackoff(); timeoutRetryCount = 0; ByteString previouslyProcessedKey = rowMerger.getLastCompletedRowKey(); operationSpan.addAnnotation("Got a response"); rowMerger.onNext(message); int rowCountInLastMessage = rowMerger.getRowCountInLastMessage(); operationSpan.addAnnotation( "Processed Response", ImmutableMap.of("rowCount", AttributeValue.longAttributeValue(rowCountInLastMessage))); totalRowsProcessed += rowCountInLastMessage; requestManager.incrementRowCount(rowCountInLastMessage); ByteString lastProcessedKey = rowMerger.getLastCompletedRowKey(); if (previouslyProcessedKey != lastProcessedKey) { updateLastFoundKey(lastProcessedKey); } else { updateLastFoundKey(message.getLastScannedRowKey()); } if (callWrapper.isAutoFlowControlEnabled()) { callWrapper.request(1); } if (resultObserver != null) { resultObserver.onNext(message); } } catch (Exception e) { setException(e); } } RetryingReadRowsOperation(
      StreamObserver<FlatRow> observer,
      RetryOptions retryOptions,
      ReadRowsRequest request,
      BigtableAsyncRpc<ReadRowsRequest, ReadRowsResponse> retryableRpc,
      CallOptions callOptions,
      ScheduledExecutorService retryExecutorService,
      Metadata originalMetadata,
      ApiClock clock); void setResultObserver(StreamObserver<ReadRowsResponse> resultObserver); @SuppressWarnings("unchecked") @Override void run(); @Override void onMessage(ReadRowsResponse message); @Override void onClose(Status status, Metadata trailers); @Override void setException(Exception exception);  }
@Test public void testEmptyResponse() { RetryingReadRowsOperation underTest = createOperation(); start(underTest); ReadRowsResponse response = ReadRowsResponse.getDefaultInstance(); underTest.onMessage(response); verify(mockFlatRowObserver, times(0)).onNext(any(FlatRow.class)); verify(mockClientCall, times(2)).request(eq(1)); verify(mockResponseObserver, times(1)).onNext(same(response)); finishOK(underTest, 0); }
@Test public void testSingleResponse() throws UnsupportedEncodingException { RetryingReadRowsOperation underTest = createOperation(); start(underTest); ByteString key = ByteString.copyFrom("SomeKey", "UTF-8"); ReadRowsResponse response = buildResponse(key); underTest.onMessage(response); verify(mockFlatRowObserver, times(1)).onNext(any(FlatRow.class)); checkRetryRequest(underTest, key, 9); verify(mockClientCall, times(2)).request(eq(1)); verify(mockResponseObserver, times(1)).onNext(same(response)); finishOK(underTest, 0); }
@Test public void testDoubleResponse() throws UnsupportedEncodingException { RetryingReadRowsOperation underTest = createOperation(); start(underTest); ByteString key1 = ByteString.copyFrom("SomeKey1", "UTF-8"); ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8"); ReadRowsResponse response = buildResponse(key1, key2); underTest.onMessage(response); verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class)); checkRetryRequest(underTest, key2, 8); verify(mockClientCall, times(2)).request(eq(1)); finishOK(underTest, 0); }
@Test public void testMultipleResponses() throws UnsupportedEncodingException { RetryingReadRowsOperation underTest = createOperation(); start(underTest); ByteString key1 = ByteString.copyFrom("SomeKey1", "UTF-8"); ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8"); underTest.onMessage(buildResponse(key1)); underTest.onMessage(buildResponse(key2)); verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class)); checkRetryRequest(underTest, key2, 8); verify(mockClientCall, times(3)).request(eq(1)); finishOK(underTest, 0); }