code: @Override
    public void onMessage(ReadRowsResponse message) {
      try {
        resetStatusBasedBackoff();
        // We've had at least one successful RPC, reset the backoff and retry counter
        timeoutRetryCount = 0;
  
        ByteString previouslyProcessedKey = rowMerger.getLastCompletedRowKey();
  
        operationSpan.addAnnotation("Got a response");
        // This may take some time. This must not block so that gRPC worker threads don't leak.
        rowMerger.onNext(message);
  
        // Add an annotation for the number of rows that were returned in the previous response.
        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) {
          // There was a full row found in the response.
          updateLastFoundKey(lastProcessedKey);
        } else {
          // The service indicates that it processed rows that did not match the filter, and will not
          // need to be reprocessed.
          updateLastFoundKey(message.getLastScannedRowKey());
        }
  
        if (callWrapper.isAutoFlowControlEnabled()) {
          callWrapper.request(1);
        }
  
        if (resultObserver != null) {
          resultObserver.onNext(message);
        }
      } catch (Exception e) {
        setException(e);
      }
    }
docs: {@inheritDoc}