| 668 | } |
| 669 | |
| 670 | Status CatalogServer::Start() { |
| 671 | TNetworkAddress subscriber_address = |
| 672 | MakeNetworkAddress(FLAGS_hostname, FLAGS_state_store_subscriber_port); |
| 673 | TNetworkAddress statestore_address = |
| 674 | MakeNetworkAddress(FLAGS_state_store_host, FLAGS_state_store_port); |
| 675 | TNetworkAddress statestore2_address = |
| 676 | MakeNetworkAddress(FLAGS_state_store_2_host, FLAGS_state_store_2_port); |
| 677 | TNetworkAddress server_address = MakeNetworkAddress(FLAGS_hostname, |
| 678 | FLAGS_catalog_service_port); |
| 679 | |
| 680 | catalog_.reset(new Catalog()); |
| 681 | #ifndef NDEBUG |
| 682 | if (FLAGS_stress_catalog_startup_delay_ms > 0) { |
| 683 | SleepForMs(FLAGS_stress_catalog_startup_delay_ms); |
| 684 | } |
| 685 | #endif |
| 686 | RETURN_IF_ERROR(Thread::Create("catalog-server", "catalog-update-gathering-thread", |
| 687 | &CatalogServer::GatherCatalogUpdatesThread, this, |
| 688 | &catalog_update_gathering_thread_)); |
| 689 | RETURN_IF_ERROR(Thread::Create("catalog-server", "catalog-first-reset-metadata-thread", |
| 690 | &CatalogServer::TriggerResetMetadata, this, &catalog_first_reset_metadata_thread_)); |
| 691 | RETURN_IF_ERROR(Thread::Create("catalog-server", "catalog-metrics-refresh-thread", |
| 692 | &CatalogServer::RefreshMetrics, this, &catalog_metrics_refresh_thread_)); |
| 693 | |
| 694 | active_catalogd_version_checker_.reset(new ActiveCatalogdVersionChecker()); |
| 695 | statestore_subscriber_.reset(new StatestoreSubscriberCatalog( |
| 696 | Substitute("catalog-server@$0", TNetworkAddressToString(server_address)), |
| 697 | subscriber_address, statestore_address, statestore2_address, metrics_, |
| 698 | protocol_version_, server_address)); |
| 699 | |
| 700 | StatestoreSubscriber::UpdateCallback cb = |
| 701 | bind<void>(mem_fn(&CatalogServer::UpdateCatalogTopicCallback), this, _1, _2); |
| 702 | // The catalogd never needs to read any entries from the topic. It only publishes |
| 703 | // entries. So, we set a prefix to some random character that we know won't be a |
| 704 | // prefix of any key. This saves a bit of network communication from the statestore |
| 705 | // back to the catalog. |
| 706 | string filter_prefix = "!"; |
| 707 | Status status = statestore_subscriber_->AddTopic(IMPALA_CATALOG_TOPIC, |
| 708 | /* is_transient=*/ false, /* populate_min_subscriber_topic_version=*/ false, |
| 709 | filter_prefix, cb); |
| 710 | if (!status.ok()) { |
| 711 | status.AddDetail("CatalogService failed to start"); |
| 712 | return status; |
| 713 | } |
| 714 | // Add callback to handle notification of updating catalogd from Statestore. |
| 715 | if (FLAGS_enable_catalogd_ha) { |
| 716 | StatestoreSubscriber::UpdateCatalogdCallback update_catalogd_cb = |
| 717 | bind<void>(mem_fn(&CatalogServer::UpdateActiveCatalogd), this, _1, _2, _3); |
| 718 | statestore_subscriber_->AddUpdateCatalogdTopic(update_catalogd_cb); |
| 719 | } |
| 720 | |
| 721 | RETURN_IF_ERROR(statestore_subscriber_->Start()); |
| 722 | if (FLAGS_force_catalogd_active && !is_active_.Load()) { |
| 723 | // If both catalogd are started with 'force_catalogd_active' as true in short time, |
| 724 | // the second election overwrite the first election. The one which registering with |
| 725 | // statestore first will be inactive. |
| 726 | LOG(WARNING) << "Could not start CatalogD as active instance"; |
| 727 | } |
no test coverage detected