Method HandleTopicUpdates

be/src/scheduling/admission-controller.cc:2197–2249

}

void AdmissionController::HandleTopicUpdates(
    const vector<TTopicItem>& topic_updates, set<string>& pool_stats_removed_nodes) {
  string topic_key_prefix;
  string topic_key_suffix;
  string pool_name;
  string topic_backend_id;
  for (const TTopicItem& item : topic_updates) {
    if (!ParseTopicKey(item.key, &topic_key_prefix, &topic_key_suffix)) continue;
    if (topic_key_prefix == TOPIC_KEY_POOL_PREFIX) {
      if (!ParsePoolTopicKey(topic_key_suffix, &pool_name, &topic_backend_id)) continue;
      // The topic entry from this subscriber is handled specially; the stats coming
      // from the statestore are likely already outdated.
      if (topic_backend_id == host_id_) continue;
      if (item.deleted) {
        GetPoolStats(pool_name)->UpdateRemoteStats(topic_backend_id, nullptr);
        pool_stats_removed_nodes.insert(topic_backend_id);
        continue;
      }
      TPoolStats remote_update;
      uint32_t len = item.value.size();
      Status status =
          DeserializeThriftMsg(reinterpret_cast<const uint8_t*>(item.value.data()), &len,
              false, &remote_update);
      if (!status.ok()) {
        VLOG_QUERY << "Error deserializing pool update with key: " << item.key;
        continue;
      }
      GetPoolStats(pool_name)->UpdateRemoteStats(topic_backend_id, &remote_update);
    } else if (topic_key_prefix == TOPIC_KEY_STAT_PREFIX) {
      topic_backend_id = topic_key_suffix;
      if (topic_backend_id == host_id_) continue;
      if (item.deleted) {
        remote_per_host_stats_.erase(topic_backend_id);
        continue;
      }
      TPerHostStatsUpdate remote_update;
      uint32_t len = item.value.size();
      Status status =
          DeserializeThriftMsg(reinterpret_cast<const uint8_t*>(item.value.data()), &len,
              false, &remote_update);
      if (!status.ok()) {
        VLOG_QUERY << "Error deserializing stats update with key: " << item.key;
        continue;
      }
      PerHostStats& stats = remote_per_host_stats_[topic_backend_id];
      for (const auto& elem : remote_update.per_host_stats) {
        stats[elem.host_addr] = elem.stats;
      }
    } else {
      VLOG_QUERY << "Invalid topic key prefix: " << topic_key_prefix;
    }
  }
}

void AdmissionController::PoolStats::UpdateAggregates(HostMemMap* host_mem_reserved) {
  const string& coord_id = parent_->host_id_;