Method ReleaseQueryBackendsLocked

be/src/scheduling/admission-controller.cc:2017–2064

}

void AdmissionController::ReleaseQueryBackendsLocked(const UniqueIdPB& query_id,
    const UniqueIdPB& coord_id, const vector<NetworkAddressPB>& host_addrs) {
  auto host_it = running_queries_.find(coord_id);
  if (host_it == running_queries_.end()) {
    // In the context of the admission control service, this may happen, e.g. if a
    // coordinator is reported as failed by the statestore but a ReleaseQuery rpc from
    // it is delayed in the network and arrives much later.
    LOG(WARNING) << "Unable to find host " << PrintId(coord_id)
                 << " to get resources to release backends for query "
                 << PrintId(query_id) << ", may have already been released.";
    return;
  }
  auto it = host_it->second.find(query_id);
  if (it == host_it->second.end()) {
    // In the context of the admission control service, this may happen, e.g. if a
    // ReleaseQueryBackends rpc is delayed in the network and arrives after the
    // ReleaseQuery rpc, so only log as a WARNING.
    LOG(WARNING) << "Unable to find resources to release backends for query "
                 << PrintId(query_id) << ", may have already been released.";
    return;
  }

  RunningQuery& running_query = it->second;
  UpdateStatsOnReleaseForBackends(query_id, running_query, host_addrs);

  // Update num_released_backends_.
  auto released_backends = num_released_backends_.find(query_id);
  if (released_backends != num_released_backends_.end()) {
    released_backends->second -= host_addrs.size();
  } else {
    // In the context of the admission control service, this may happen, e.g. if a
    // ReleaseQueryBackends rpc is delayed in the network and arrives after the
    // ReleaseQuery rpc, so only log as a WARNING.
    string err_msg = Substitute(
        "Unable to find num released backends for query $0", PrintId(query_id));
    LOG(WARNING) << err_msg;
  }

  if (VLOG_IS_ON(2)) {
    stringstream ss;
    ss << "Released query backend(s) ";
    for (const auto& host_addr : host_addrs) ss << host_addr << " ";
    ss << "for query id=" << PrintId(query_id) << " "
       << GetPoolStats(running_query.request_pool)->DebugString();
    VLOG(2) << ss.str();
  }
  pending_dequeue_ = true;
}

vector<UniqueIdPB> AdmissionController::CleanupQueriesForHost(
    const UniqueIdPB& coord_id, const std::unordered_set<UniqueIdPB>& query_ids) {