thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-25 17:12:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

More robust ClusterDiscovery::updateCluster

Browse Source
This commit is contained in:
vdimir 2022-08-17 14:46:06 +00:00
parent 89dca02e1d
commit 7305158a58
No known key found for this signature in database
GPG Key ID: 6EE4CE2BEDC51862
1 changed files with 17 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/Interpreters/ClusterDiscovery.cpp
View File

@ -266,6 +266,21 @@ bool ClusterDiscovery::updateCluster(ClusterInfo & cluster_info)
Strings node_uuids = getNodeNames(zk, cluster_info.zk_root, cluster_info.name, &start_version, false);
auto & nodes_info = cluster_info.nodes_info;
auto on_exit = [this, start_version, &zk, &cluster_info, &nodes_info]()
{
/// in case of successful update we still need to check if configuration of cluster still valid and also set watch callback
int current_version;
getNodeNames(zk, cluster_info.zk_root, cluster_info.name, &current_version, true);
if (current_version != start_version)
{
LOG_DEBUG(log, "Cluster '{}' configuration changed during update", cluster_info.name);
nodes_info.clear();
return false;
}
return true;
};
if (!cluster_info.current_node_is_observer && !contains(node_uuids, current_node_name))
{
LOG_ERROR(log, "Can't find current node in cluster '{}', will register again", cluster_info.name);
@ -277,7 +292,7 @@ bool ClusterDiscovery::updateCluster(ClusterInfo & cluster_info)
if (!needUpdate(node_uuids, nodes_info))
{
LOG_DEBUG(log, "No update required for cluster '{}'", cluster_info.name);
return true;
return on_exit();
}
nodes_info = getNodes(zk, cluster_info.zk_root, node_uuids);
@ -287,15 +302,8 @@ bool ClusterDiscovery::updateCluster(ClusterInfo & cluster_info)
return false;
}
int current_version;
getNodeNames(zk, cluster_info.zk_root, cluster_info.name, &current_version, true);
if (current_version != start_version)
{
LOG_DEBUG(log, "Cluster '{}' configuration changed during update", cluster_info.name);
nodes_info.clear();
if (bool ok = on_exit(); !ok)
return false;
}
LOG_DEBUG(log, "Updating system.clusters record for '{}' with {} nodes", cluster_info.name, cluster_info.nodes_info.size());