Method SendTopicUpdate

be/src/statestore/statestore.cc:1058–1173

}

Status Statestore::SendTopicUpdate(Subscriber* subscriber, UpdateKind update_kind,
    bool* update_skipped) {
  if (!IsActive()) {
    // Don't send topic update if the statestored is not active.
    return Status::OK();
  } else if (FLAGS_enable_catalogd_ha && subscriber->IsCatalogd()
      && !catalog_manager_.IsActiveCatalogd(subscriber->id())) {
    // Don't send topic update to inactive catalogd.
    VLOG(3) << "Skip sending topic update to inactive catalogd";
    return Status::OK();
  }

  // Time any successful RPCs (i.e. those for which UpdateState() completed, even though
  // it may have returned an error.)
  MonotonicStopWatch sw;
  sw.Start();

  // First thing: make a list of updates to send
  TUpdateStateRequest update_state_request;
  GatherTopicUpdates(*subscriber, update_kind, &update_state_request);
  // 'subscriber' may not be subscribed to any updates of 'update_kind'.
  if (update_state_request.topic_deltas.empty()) {
    *update_skipped = false;
    return Status::OK();
  }

  // Set the expected registration ID, so that the subscriber can reject this update if
  // they have moved on to a new registration instance.
  update_state_request.__set_registration_id(subscriber->registration_id());
  update_state_request.__set_statestore_id(statestore_id_);

  // Second: try and send it
  Status status;
  StatestoreSubscriberConn client(update_state_client_cache_.get(),
      subscriber->network_address(), &status);
  RETURN_IF_ERROR(status);

  TUpdateStateResponse response;
  RETURN_IF_ERROR(client.DoRpc(
      &StatestoreSubscriberClientWrapper::UpdateState, update_state_request, &response));

  StatsMetric<double>* update_duration_metric =
      update_kind == UpdateKind::PRIORITY_TOPIC_UPDATE ?
      priority_topic_update_duration_metric_ : topic_update_duration_metric_;
  status = StatusFromThrift(response.status);
  if (!status.ok()) {
    update_duration_metric->Update(sw.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
    return status;
  }

  *update_skipped = (response.__isset.skipped && response.skipped);
  if (*update_skipped) {
    // The subscriber skipped processing this update. We don't consider this a failure
    // - subscribers can decide what they do with any update - so, return OK and set
    // update_skipped so the caller can compensate.
    update_duration_metric->Update(sw.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
    return Status::OK();
  }