docs: @Override
    public ReadRowsRequest getResumeRequest(ReadRowsRequest originalRequest) {
      // An empty lastKey means that we have not successfully read the first row,
      // so resume with the original request object.
      if (lastKey.isEmpty()) {
        return originalRequest;
      }
  
      RowSet remaining = RowSetUtil.split(originalRequest.getRows(), lastKey).getRight();
  
      // Edge case: retrying a fulfilled request.
      // A fulfilled request is one that has had all of its row keys and ranges fulfilled, or if it
      // had a row limit, has seen enough rows. These requests are replaced with a marker request that
      // will be handled by ReadRowsRetryCompletedCallable. See docs in ReadRowsRetryCompletedCallable
      // for more details.
      if (remaining == null
          || (originalRequest.getRowsLimit() > 0 && originalRequest.getRowsLimit() == numProcessed)) {
        return ReadRowsRetryCompletedCallable.FULFILLED_REQUEST_MARKER;
      }
  
      Builder builder = originalRequest.toBuilder().setRows(remaining);
  
      if (originalRequest.getRowsLimit() > 0) {
        Preconditions.checkState(
            originalRequest.getRowsLimit() > numProcessed,
            "Detected too many rows for the current row limit during a retry.");
        builder.setRowsLimit(originalRequest.getRowsLimit() - numProcessed);
      }
  
      return builder.build();
    }