Method GatherCatalogUpdatesThread

be/src/catalog/catalog-server.cc:1066–1116

}

[[noreturn]] void CatalogServer::GatherCatalogUpdatesThread() {
  // catalog_topic_mode=minimal does not require initial reset to happen ahead of catalog
  // update gathering because coordinator will request metadata on-demand.
  bool require_initial_reset = FLAGS_catalog_topic_mode != "minimal";
  while (true) {
    unique_lock<mutex> unique_lock(catalog_lock_);
    // Protect against spurious wake-ups by checking the value of topic_updates_ready_.
    // It is only safe to continue on and update the shared pending_topic_updates_
    // when topic_updates_ready_ is false, otherwise we may be in the middle of
    // processing a heartbeat.
    // If require_initial_reset is True, this thread need to let TriggerResetMetadata
    // thread to proceed first.
    while (topic_updates_ready_
        || (require_initial_reset && !triggered_pending_reset_.Load())) {
      if (!triggered_pending_reset_.Load()) {
        // Wake up TriggerResetMetadata thread.
        catalog_update_cv_.NotifyOne();
      }
      catalog_update_cv_.Wait(unique_lock);
    }

    MonotonicStopWatch sw;
    sw.Start();

    // Clear any pending topic updates. They will have been processed by the heartbeat
    // thread by the time we make it here.
    pending_topic_updates_.clear();

    long current_catalog_version;
    Status status = catalog_->GetCatalogVersion(&current_catalog_version);
    if (!status.ok()) {
      LOG(ERROR) << status.GetDetail();
    } else if (current_catalog_version != last_sent_catalog_version_) {
      // If there has been a change since the last time the catalog was queried,
      // call into the Catalog to find out what has changed.
      VLOG(2) << "Catalog version changed from " << last_sent_catalog_version_ << " to "
              << current_catalog_version << ". Gathering catalog delta.";
      TGetCatalogDeltaResponse resp;
      status = catalog_->GetCatalogDelta(this, last_sent_catalog_version_, &resp);
      if (!status.ok()) {
        LOG(ERROR) << status.GetDetail();
      } else {
        catalog_objects_max_version_ = resp.max_catalog_version;
      }
    }

    topic_processing_time_metric_->Update(sw.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
    topic_updates_ready_ = true;
    if (!triggered_pending_reset_.Load()) catalog_update_cv_.NotifyOne();
  }
}

[[noreturn]] void CatalogServer::RefreshMetrics() {
  while (true) {