Method CloseSessionInternal

be/src/service/impala-server.cc:2044–2107

}

Status ImpalaServer::CloseSessionInternal(const TUniqueId& session_id,
    const SecretArg& secret, bool ignore_if_absent) {
  DCHECK(secret.is_session_secret());
  VLOG_QUERY << "Closing session: " << PrintId(session_id);

  // Find the session_state and remove it from the map.
  shared_ptr<SessionState> session_state;
  {
    lock_guard<mutex> l(session_state_map_lock_);
    SessionStateMap::iterator entry = session_state_map_.find(session_id);
    if (entry == session_state_map_.end() || !secret.Validate(entry->second->secret)) {
      if (ignore_if_absent) {
        return Status::OK();
      } else {
        if (entry != session_state_map_.end()) {
          // Log invalid attempts to connect. Be careful not to log secret.
          VLOG(1) << "Client tried to connect to session " << PrintId(session_id)
                  << " with invalid secret.";
        }
        string err_msg = Substitute("Invalid session id: $0", PrintId(session_id));
        VLOG(1) << "CloseSessionInternal(): " << err_msg;
        return Status::Expected(err_msg);
      }
    }
    session_state = entry->second;
    session_state_map_.erase(session_id);
  }
  DCHECK(session_state != nullptr);
  if (session_state->session_type == TSessionType::BEESWAX) {
    ImpaladMetrics::IMPALA_SERVER_NUM_OPEN_BEESWAX_SESSIONS->Increment(-1L);
  } else {
    ImpaladMetrics::IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS->Increment(-1L);
    DecrementSessionCount(session_state->connected_user);
  }
  unordered_set<TUniqueId> inflight_queries;
  vector<TUniqueId> interrupted_queries;
  {
    lock_guard<mutex> l(session_state->lock);
    DCHECK(!session_state->closed);
    session_state->closed = true;
    // Since closed is true, no more queries will be added to the inflight list.
    inflight_queries.insert(session_state->inflight_queries.begin(),
        session_state->inflight_queries.end());
    interrupted_queries.swap(session_state->interrupted_queries);
  }
  // Unregister all open queries from this session.
  Status status = Status::Expected("Session closed");
  for (const TUniqueId& query_id: inflight_queries) {
    // TODO: deal with an error status
    UnregisterQueryDiscardResult(query_id, &status);
  }
  {
    lock_guard<mutex> l(interrupted_query_statuses_lock_);
    for (const TUniqueId& query_id: interrupted_queries) {
      interrupted_query_statuses_.erase(query_id);
    }
  }
  // Reconfigure the poll period of session_maintenance_thread_ if necessary.