Method WorkloadManagementWorker

be/src/service/workload-management-worker.cc:636–822

} // function ImpalaServer::IsWorkloadManagementShuttingDown

void ImpalaServer::WorkloadManagementWorker(const Version& target_schema_version) {
  {
    lock_guard<mutex> l(workload_mgmt_state_mu_);
    workload_mgmt_state_ = WorkloadManagementState::STARTING;
  }

  {
    lock_guard<mutex> l(workload_mgmt_state_mu_);
    workload_mgmt_state_ = WorkloadManagementState::STARTED;
  }

  {
    lock_guard<mutex> l(workload_mgmt_state_mu_);
    // This condition will evaluate to false only if a clean shutdown was initiated while
    // the previous function was running.
    if (LIKELY(workload_mgmt_state_ == WorkloadManagementState::STARTED)) {
      workload_mgmt_state_ = WorkloadManagementState::RUNNING;
    } else {
      LOG(INFO) << "Not starting workload management processing thread because "
                << "coordinator shutdown was initiated.";
      return; // Note: early return
    }
  }

  // Non-values portion of the sql DML to insert records into the completed queries
  // tables. This portion of the statement is constant and thus is only generated once.
  const string log_table_name = QueryLogTableName(true);
  const string insert_dml_prefix = _dmlPrefix(log_table_name, target_schema_version);
  VLOG(2) << "Workload Management insert sql prefix: " << insert_dml_prefix;

  // Setup default query options that will be provided on all queries that insert rows
  // into the completed queries table.
  InternalServer::QueryOptionMap insert_query_opts;

  insert_query_opts[TImpalaQueryOptions::TIMEZONE] = "UTC";
  insert_query_opts[TImpalaQueryOptions::QUERY_TIMEOUT_S] = std::to_string(
      FLAGS_query_log_write_timeout_s < 1 ? FLAGS_query_log_write_interval_s :
                                            FLAGS_query_log_write_timeout_s);
  if (!FLAGS_query_log_request_pool.empty()) {
    insert_query_opts[TImpalaQueryOptions::REQUEST_POOL] = FLAGS_query_log_request_pool;
  }
  insert_query_opts[TImpalaQueryOptions::FETCH_ROWS_TIMEOUT_MS] = "0";
  insert_query_opts[TImpalaQueryOptions::EXEC_TIME_LIMIT_S] =
      std::to_string(FLAGS_query_log_dml_exec_timeout_s);
  if (!FLAGS_debug_actions.empty()) {
    insert_query_opts[TImpalaQueryOptions::DEBUG_ACTION] = FLAGS_debug_actions;
  }
  // Hide analyzed query since it can be prohibitively long.
  insert_query_opts[TImpalaQueryOptions::HIDE_ANALYZED_QUERY] = "true";

  while (true) {
    // Exit this thread if a shutdown was initiated.
    if (IsWorkloadManagementShuttingDown()) {
      _completed_queries_shutdown_cv.notify_all();
      return; // Note: early return
    }

    // Sleep this thread until it is time to process queued completed queries or the