Add ability to sync ZooKeeper before reloading the config

Otherwise it is not deterministic wether ClickHouse will see this
change, i.e. if you created znode just before `SYSTEM RELOAD CONFIG` it
is not 100% that the ClickHouse will see it.

Refs: apache/zookeeper@8ee17324ce
Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

programs/server/Server.cpp

@@ -1245,9 +1245,14 @@ try
     if (loaded_config.has_zk_includes)
     {
         auto old_configuration = loaded_config.configuration;
+        bool config_reload_sync_zookeeper = old_configuration->getBool("config_reload_sync_zookeeper", false);
         ConfigProcessor config_processor(config_path);
         loaded_config = config_processor.loadConfigWithZooKeeperIncludes(
-            main_config_zk_node_cache, main_config_zk_changed_event, /* fallback_to_preprocessed = */ true);
+            main_config_zk_node_cache,
+            main_config_zk_changed_event,
+            /*fallback_to_preprocessed =*/ true,
+            /*is_config_changed=*/ true,
+            config_reload_sync_zookeeper);
         config_processor.savePreprocessedConfig(loaded_config, path_str);
         config().removeConfiguration(old_configuration.get());
         config().add(loaded_config.configuration.duplicate(), PRIO_DEFAULT, false);

src/Common/Config/ConfigProcessor.cpp

@@ -811,13 +811,17 @@ ConfigProcessor::LoadedConfig ConfigProcessor::loadConfigWithZooKeeperIncludes(
     zkutil::ZooKeeperNodeCache & zk_node_cache,
     const zkutil::EventPtr & zk_changed_event,
     bool fallback_to_preprocessed,
-    bool is_config_changed)
+    bool is_config_changed,
+    bool sync_zookeeper)
 {
     XMLDocumentPtr config_xml;
     bool has_zk_includes;
     bool processed_successfully = false;
     try
     {
+        if (sync_zookeeper)
+            zk_node_cache.sync();
+
         config_xml = processConfig(&has_zk_includes, &zk_node_cache, zk_changed_event, is_config_changed);
         processed_successfully = true;
     }

src/Common/Config/ConfigProcessor.h

@@ -96,8 +96,9 @@ public:
     LoadedConfig loadConfigWithZooKeeperIncludes(
         zkutil::ZooKeeperNodeCache & zk_node_cache,
         const zkutil::EventPtr & zk_changed_event,
-        bool fallback_to_preprocessed = false,
-        bool is_config_changed = true);
+        bool fallback_to_preprocessed,
+        bool is_config_changed,
+        bool sync_zookeeper);
 
     /// Save preprocessed config to specified directory.
     /// If preprocessed_dir is empty - calculate from loaded_config.path + /preprocessed_configs/

src/Common/Config/ConfigReloader.cpp

@@ -123,8 +123,15 @@ std::optional<ConfigProcessor::LoadedConfig> ConfigReloader::reloadIfNewer(bool
         {
             loaded_config = config_processor.loadConfig(/* allow_zk_includes = */ true, is_config_changed);
             if (loaded_config.has_zk_includes)
+            {
+                bool config_reload_sync_zookeeper = loaded_config.configuration->getBool("config_reload_sync_zookeeper", false);
                 loaded_config = config_processor.loadConfigWithZooKeeperIncludes(
-                    zk_node_cache, zk_changed_event, fallback_to_preprocessed, is_config_changed);
+                    zk_node_cache,
+                    zk_changed_event,
+                    fallback_to_preprocessed,
+                    is_config_changed,
+                    config_reload_sync_zookeeper);
+            }
         }
         catch (const Coordination::Exception & e)
         {

src/Common/ZooKeeper/ZooKeeperNodeCache.cpp

@@ -105,4 +105,9 @@ ZooKeeperNodeCache::ZNode ZooKeeperNodeCache::get(const std::string & path, Coor
     return result;
 }
 
+void ZooKeeperNodeCache::sync()
+{
+    get_zookeeper()->sync("/");
+}
+
 }

src/Common/ZooKeeper/ZooKeeperNodeCache.h

@@ -46,6 +46,8 @@ public:
     ZNode get(const std::string & path, EventPtr watch_event);
     ZNode get(const std::string & path, Coordination::WatchCallback watch_callback);
 
+    void sync();
+
 private:
     GetZooKeeper get_zookeeper;
 

src/Core/ServerSettings.cpp

@@ -195,6 +195,7 @@ namespace DB
     DECLARE(String, mutation_workload, "default", "Name of workload to be used to access resources for all mutations (may be overridden by a merge tree setting)", 0) \
     DECLARE(Bool, prepare_system_log_tables_on_startup, false, "If true, ClickHouse creates all configured `system.*_log` tables before the startup. It can be helpful if some startup scripts depend on these tables.", 0) \
     DECLARE(UInt64, config_reload_interval_ms, 2000, "How often clickhouse will reload config and check for new changes", 0) \
+    DECLARE(Bool, config_reload_sync_zookeeper, false, "Sync ZooKeeper before reloading config", 0) \
     DECLARE(UInt64, memory_worker_period_ms, 0, "Tick period of background memory worker which corrects memory tracker memory usages and cleans up unused pages during higher memory usage. If set to 0, default value will be used depending on the memory usage source", 0) \
     DECLARE(Bool, disable_insertion_and_mutation, false, "Disable all insert/alter/delete queries. This setting will be enabled if someone needs read-only nodes to prevent insertion and mutation affect reading performance.", 0) \
     DECLARE(UInt64, parts_kill_delay_period, 30, "Period to completely remove parts for SharedMergeTree. Only available in ClickHouse Cloud", 0) \

tests/integration/test_config_substitutions/configs/config_zk_include_test.xml

@@ -1,5 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <clickhouse>
+  <config_reload_sync_zookeeper>true</config_reload_sync_zookeeper>
+
   <background_pool_size>44</background_pool_size>
   <merge_tree>
     <include from_zk="/merge_max_block_size" merge="true"/>