Method ProcessScanToken

be/src/exec/kudu/kudu-scan-node.cc:204–223

}

Status KuduScanNode::ProcessScanToken(KuduScanner* scanner, const string& scan_token) {
  bool eos;
  RETURN_IF_ERROR(scanner->OpenNextScanToken(scan_token, &eos));
  if (eos) return Status::OK();
  while (!eos && !done_.Load()) {
    unique_ptr<RowBatch> row_batch = std::make_unique<RowBatch>(row_desc(),
        runtime_state_->batch_size(), mem_tracker());
    RETURN_IF_ERROR(scanner->GetNext(row_batch.get(), &eos));
    while (!done_.Load()) {
      scanner->KeepKuduScannerAlive();
      if (thread_state_.EnqueueBatchWithTimeout(&row_batch, 1000000)) {
        break;
      }
      // Make sure that we still own the RowBatch if BlockingPutWithTimeout() timed out.
      DCHECK(row_batch != nullptr);
    }
  }
  if (eos) scan_ranges_complete_counter_->Add(1);
  return Status::OK();
}

void KuduScanNode::RunScannerThread(
    bool first_thread, const string& name, const string* initial_token) {