| 861 | } |
| 862 | |
| 863 | void CatalogServer::UpdateActiveCatalogd(bool is_registration_reply, |
| 864 | int64_t active_catalogd_version, const TCatalogRegistration& catalogd_registration) { |
| 865 | lock_guard<mutex> meta_lock(ha_transition_lock_); |
| 866 | unique_lock<mutex> l(catalog_lock_); |
| 867 | if (!active_catalogd_version_checker_->CheckActiveCatalogdVersion( |
| 868 | is_registration_reply, active_catalogd_version)) { |
| 869 | return; |
| 870 | } |
| 871 | if (catalogd_registration.address.hostname.empty() |
| 872 | || catalogd_registration.address.port == 0) { |
| 873 | return; |
| 874 | } |
| 875 | LOG(INFO) << "Get notification of active catalogd: " |
| 876 | << TNetworkAddressToString(catalogd_registration.address); |
| 877 | bool is_matching = (catalogd_registration.address.hostname == FLAGS_hostname |
| 878 | && catalogd_registration.address.port == FLAGS_catalog_service_port); |
| 879 | if (is_matching) { |
| 880 | if (!is_active_.Load()) { |
| 881 | is_active_.Store(true); |
| 882 | active_status_metric_->SetValue(true); |
| 883 | num_ha_active_status_change_metric_->Increment(1); |
| 884 | // Reset last_sent_catalog_version_ when the catalogd become active. This will |
| 885 | // lead to non-delta catalog update for next IMPALA_CATALOG_TOPIC which also |
| 886 | // instruct the statestore to clear all entries for the catalog update topic. |
| 887 | last_sent_catalog_version_ = 0; |
| 888 | // Regenerate Catalog Service ID. |
| 889 | catalog_->RegenerateServiceId(); |
| 890 | // Clear pending topic updates. |
| 891 | pending_topic_updates_.clear(); |
| 892 | |
| 893 | if (FLAGS_catalogd_ha_reset_metadata_on_failover) { |
| 894 | MarkPendingMetadataReset(l); |
| 895 | } else { |
| 896 | // Refresh DataSource objects when the catalogd becomes active. |
| 897 | Status status = catalog_->RefreshDataSources(); |
| 898 | if (!status.ok()) { |
| 899 | LOG(ERROR) << "Failed to refresh data sources triggered by catalogd failover."; |
| 900 | } |
| 901 | // If HA state has been determined, this is a failover. Apply pending HMS events |
| 902 | // to avoid stale metadata. Note that only HMS events before the failover happens |
| 903 | // need to be applied. HMS events after that are OK to be applied later since they |
| 904 | // are not applied in the previous active catalogd as well. |
| 905 | if (is_ha_determined_) WaitUntilHmsEventsSynced(l); |
| 906 | } |
| 907 | // Signal the catalog update gathering thread to start. |
| 908 | topic_updates_ready_ = false; |
| 909 | catalog_update_cv_.NotifyOne(); |
| 910 | LOG(INFO) << "This catalogd instance is changed to active status"; |
| 911 | } |
| 912 | } else { |
| 913 | if (is_active_.Load()) { |
| 914 | is_active_.Store(false); |
| 915 | active_status_metric_->SetValue(false); |
| 916 | num_ha_active_status_change_metric_->Increment(1); |
| 917 | LOG(INFO) << "This catalogd instance is changed to inactive status. " |
| 918 | << "Current active catalogd: " |
| 919 | << TNetworkAddressToString(catalogd_registration.address) |
| 920 | << ", active_catalogd_version: " |
nothing calls this directly
no test coverage detected