Method ExecCompleteCb

be/src/runtime/coordinator-backend-state.cc:242–289

}

void Coordinator::BackendState::ExecCompleteCb(
    TypedCountingBarrier<Status>* exec_status_barrier, int64_t start_ms) {
  {
    lock_guard<mutex> l(lock_);
    exec_rpc_status_ = exec_rpc_controller_.status();

    Status complete_cb_debug_status =
        DebugAction(exec_params_.query_options(), "IMPALA_MISS_EXEC_COMPLETE_CB");
    if (UNLIKELY(exec_rpc_status_.ok() && !complete_cb_debug_status.ok())) {
      // Simulate the missing of callback for successful RPC.
      LOG(ERROR) << "Debug action: missing ExecComplete callback";
      return;
    }

    rpc_latency_ = MonotonicMillis() - start_ms;

    if (!exec_rpc_status_.ok()) {
      // Return CANCELLED instead of ABORTED if the RPC is cancelled.
      if (cancel_exec_rpc_ && exec_rpc_status_.IsAborted()) {
        LOG(ERROR) << "ExecQueryFInstances rpc query_id=" << PrintId(query_id_)
                   << " was aborted by cancellation";
        status_ = Status::CANCELLED;
        exec_done_ = true;
        exec_status_barrier->NotifyRemaining(status_);
      } else {
        SetExecError(
            FromKuduStatus(exec_rpc_status_, "Exec() rpc failed"), exec_status_barrier);
      }
      goto done;
    }

    Status exec_status = StatusFromProto(exec_response_.status());
    if (!exec_status.ok()) {
      SetExecError(exec_status, exec_status_barrier);
      goto done;
    }

    for (const auto& entry : instance_stats_map_) entry.second->stopwatch_.Start();
    VLOG_FILE << "rpc succeeded: ExecQueryFInstances query_id=" << PrintId(query_id_);
    exec_done_ = true;
    last_report_time_ms_ = GenerateReportTimestamp();
    exec_status_barrier->Notify(Status::OK());
  }
done:
  // Notify after releasing 'lock_' so that we don't wake up a thread just to have it
  // immediately block again.
  exec_done_cv_.NotifyAll();
}

void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
    const FilterRoutingTable& filter_routing_table,