Merge pull request #71979 from ClickHouse/backport/24.3/69274

Backport #69274 to 24.3: fix `metadata_version` in ZooKeeper
This commit is contained in:
Alexander Tokmakov 2024-11-19 16:27:09 +01:00 committed by GitHub
commit aba2872e49
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 74 additions and 3 deletions

View File

@ -14,6 +14,7 @@ namespace ErrorCodes
{ {
extern const int SUPPORT_IS_DISABLED; extern const int SUPPORT_IS_DISABLED;
extern const int REPLICA_STATUS_CHANGED; extern const int REPLICA_STATUS_CHANGED;
extern const int LOGICAL_ERROR;
} }
ReplicatedMergeTreeAttachThread::ReplicatedMergeTreeAttachThread(StorageReplicatedMergeTree & storage_) ReplicatedMergeTreeAttachThread::ReplicatedMergeTreeAttachThread(StorageReplicatedMergeTree & storage_)
@ -116,6 +117,67 @@ void ReplicatedMergeTreeAttachThread::checkHasReplicaMetadataInZooKeeper(const z
} }
} }
Int32 ReplicatedMergeTreeAttachThread::fixReplicaMetadataVersionIfNeeded(zkutil::ZooKeeperPtr zookeeper)
{
const String & zookeeper_path = storage.zookeeper_path;
const String & replica_path = storage.replica_path;
const bool replica_readonly = storage.is_readonly;
for (size_t i = 0; i != 2; ++i)
{
String replica_metadata_version_str;
const bool replica_metadata_version_exists = zookeeper->tryGet(replica_path + "/metadata_version", replica_metadata_version_str);
if (!replica_metadata_version_exists)
return -1;
const Int32 metadata_version = parse<Int32>(replica_metadata_version_str);
if (metadata_version != 0 || replica_readonly)
{
/// No need to fix anything
return metadata_version;
}
Coordination::Stat stat;
zookeeper->get(fs::path(zookeeper_path) / "metadata", &stat);
if (stat.version == 0)
{
/// No need to fix anything
return metadata_version;
}
ReplicatedMergeTreeQueue & queue = storage.queue;
queue.pullLogsToQueue(zookeeper);
if (queue.getStatus().metadata_alters_in_queue != 0)
{
LOG_DEBUG(log, "No need to update metadata_version as there are ALTER_METADATA entries in the queue");
return metadata_version;
}
const Coordination::Requests ops = {
zkutil::makeSetRequest(fs::path(replica_path) / "metadata_version", std::to_string(stat.version), 0),
zkutil::makeCheckRequest(fs::path(zookeeper_path) / "metadata", stat.version),
};
Coordination::Responses ops_responses;
const auto code = zookeeper->tryMulti(ops, ops_responses);
if (code == Coordination::Error::ZOK)
{
LOG_DEBUG(log, "Successfully set metadata_version to {}", stat.version);
return stat.version;
}
if (code != Coordination::Error::ZBADVERSION)
{
throw zkutil::KeeperException(code);
}
}
/// Second attempt is only possible if metadata_version != 0 or metadata.version changed during the first attempt.
/// If metadata_version != 0, on second attempt we will return the new metadata_version.
/// If metadata.version changed, on second attempt we will either get metadata_version != 0 and return the new metadata_version or we will get metadata_alters_in_queue != 0 and return 0.
/// Either way, on second attempt this method should return.
throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to fix replica metadata_version in ZooKeeper after two attempts");
}
void ReplicatedMergeTreeAttachThread::runImpl() void ReplicatedMergeTreeAttachThread::runImpl()
{ {
storage.setZooKeeper(); storage.setZooKeeper();
@ -159,11 +221,11 @@ void ReplicatedMergeTreeAttachThread::runImpl()
/// Just in case it was not removed earlier due to connection loss /// Just in case it was not removed earlier due to connection loss
zookeeper->tryRemove(replica_path + "/flags/force_restore_data"); zookeeper->tryRemove(replica_path + "/flags/force_restore_data");
String replica_metadata_version; const Int32 replica_metadata_version = fixReplicaMetadataVersionIfNeeded(zookeeper);
const bool replica_metadata_version_exists = zookeeper->tryGet(replica_path + "/metadata_version", replica_metadata_version); const bool replica_metadata_version_exists = replica_metadata_version != -1;
if (replica_metadata_version_exists) if (replica_metadata_version_exists)
{ {
storage.setInMemoryMetadata(metadata_snapshot->withMetadataVersion(parse<int>(replica_metadata_version))); storage.setInMemoryMetadata(metadata_snapshot->withMetadataVersion(replica_metadata_version));
} }
else else
{ {

View File

@ -48,6 +48,8 @@ private:
void runImpl(); void runImpl();
void finalizeInitialization(); void finalizeInitialization();
Int32 fixReplicaMetadataVersionIfNeeded(zkutil::ZooKeeperPtr zookeeper);
}; };
} }

View File

@ -2099,6 +2099,7 @@ ReplicatedMergeTreeQueue::Status ReplicatedMergeTreeQueue::getStatus() const
res.inserts_in_queue = 0; res.inserts_in_queue = 0;
res.merges_in_queue = 0; res.merges_in_queue = 0;
res.part_mutations_in_queue = 0; res.part_mutations_in_queue = 0;
res.metadata_alters_in_queue = 0;
res.queue_oldest_time = 0; res.queue_oldest_time = 0;
res.inserts_oldest_time = 0; res.inserts_oldest_time = 0;
res.merges_oldest_time = 0; res.merges_oldest_time = 0;
@ -2141,6 +2142,11 @@ ReplicatedMergeTreeQueue::Status ReplicatedMergeTreeQueue::getStatus() const
res.oldest_part_to_mutate_to = entry->new_part_name; res.oldest_part_to_mutate_to = entry->new_part_name;
} }
} }
if (entry->type == LogEntry::ALTER_METADATA)
{
++res.metadata_alters_in_queue;
}
} }
return res; return res;

View File

@ -451,6 +451,7 @@ public:
UInt32 inserts_in_queue; UInt32 inserts_in_queue;
UInt32 merges_in_queue; UInt32 merges_in_queue;
UInt32 part_mutations_in_queue; UInt32 part_mutations_in_queue;
UInt32 metadata_alters_in_queue;
UInt32 queue_oldest_time; UInt32 queue_oldest_time;
UInt32 inserts_oldest_time; UInt32 inserts_oldest_time;
UInt32 merges_oldest_time; UInt32 merges_oldest_time;