@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); }
@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 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 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); }
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); }
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);  }