code: @Override
    public final void onNext(ReadRowsResponse readRowsResponse) {
      if (complete) {
        onError(new IllegalStateException("Adding partialRow after completion"));
        return;
      }
      int rowsProcessed = 0;
      for (int i = 0; i < readRowsResponse.getChunksCount(); i++) {
        try {
          CellChunk chunk = readRowsResponse.getChunks(i);
          state.validateChunk(rowInProgress, lastCompletedRowKey, chunk);
          if (chunk.getResetRow()) {
            rowInProgress = null;
            state = RowMergerState.NewRow;
            continue;
          }
          if(state == RowMergerState.NewRow) {
            rowInProgress = new RowInProgress();
            rowInProgress.updateCurrentKey(chunk);
          } else if (state == RowMergerState.RowInProgress) {
            rowInProgress.updateCurrentKey(chunk);
          }
          if (chunk.getValueSize() > 0) {
            rowInProgress.addPartialCellChunk(chunk);
            state = RowMergerState.CellInProgress;
          } else if (rowInProgress.hasChunkInProgess()) {
            rowInProgress.addPartialCellChunk(chunk);
            rowInProgress.completeMultiChunkCell();
            state = RowMergerState.RowInProgress;
          } else {
            rowInProgress.addFullChunk(chunk);
            state = RowMergerState.RowInProgress;
          }
  
          if (chunk.getCommitRow()) {
            observer.onNext(rowInProgress.buildRow());
            lastCompletedRowKey = rowInProgress.getRowKey();
            state = RowMergerState.NewRow;
            rowInProgress = null;
            rowsProcessed++;
          }
        } catch (Throwable e) {
          onError(e);
          return;
        }
      }
      this.rowCountInLastMessage = rowsProcessed;
    }
docs: {@inheritDoc}