Backport #84463 to 25.5: S3Queue ordered mode fix: quit earlier if shutdown

robot-clickhouse · robot-clickhouse · commit 5ab1587800ee · 2025-07-28T14:14:12.000Z
diff --git a/src/Storages/ObjectStorageQueue/ObjectStorageQueueOrderedFileMetadata.cpp b/src/Storages/ObjectStorageQueue/ObjectStorageQueueOrderedFileMetadata.cpp
@@ -202,7 +202,9 @@ ObjectStorageQueueOrderedFileMetadata::BucketHolderPtr ObjectStorageQueueOrdered
 
     const auto processor_info = getProcessorInfo(processor);
 
-    while (true)
+    const size_t max_num_tries = 1000;
+    Coordination::Error code;
+    for (size_t i = 0; i < max_num_tries; ++i)
     {
         Coordination::Requests requests;
 
@@ -225,7 +227,7 @@ ObjectStorageQueueOrderedFileMetadata::BucketHolderPtr ObjectStorageQueueOrdered
         requests.push_back(zkutil::makeSetRequest(bucket_lock_id_path, processor_info, -1));
 
         Coordination::Responses responses;
-        const auto code = zk_client->tryMulti(requests, responses);
+        code = zk_client->tryMulti(requests, responses);
         if (code == Coordination::Error::ZOK)
         {
             const auto & set_response = dynamic_cast<const Coordination::SetResponse &>(*responses.back());
@@ -259,6 +261,11 @@ ObjectStorageQueueOrderedFileMetadata::BucketHolderPtr ObjectStorageQueueOrdered
         LOG_INFO(log_, "Bucket lock id path was probably created or removed "
                  "while acquiring the bucket (error code: {}), will retry", code);
     }
+
+    throw Exception(
+        ErrorCodes::LOGICAL_ERROR,
+        "Failed to set file processing within {} retries, last error: {}",
+        max_num_tries, code);
 }
 
 std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorageQueueOrderedFileMetadata::setProcessingImpl()
@@ -267,7 +274,9 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
     processing_id = node_metadata.processing_id = getRandomASCIIString(10);
     auto processor_info = getProcessorInfo(processing_id.value());
 
-    while (true)
+    const size_t max_num_tries = 1000;
+    Coordination::Error code;
+    for (size_t i = 0; i < max_num_tries; ++i)
     {
         std::optional<NodeMetadata> processed_node;
         Coordination::Stat processed_node_stat;
@@ -277,10 +286,10 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
             std::vector<std::string> paths{processed_node_path, failed_node_path};
             auto responses = zk_client->tryGet(paths);
 
-            auto check_code = [this](auto code)
+            auto check_code = [this](auto code_)
             {
-                if (!(code == Coordination::Error::ZOK || code == Coordination::Error::ZNONODE))
-                    throw zkutil::KeeperException::fromPath(code, path);
+                if (!(code_ == Coordination::Error::ZOK || code_ == Coordination::Error::ZNONODE))
+                    throw zkutil::KeeperException::fromPath(code_, path);
             };
             check_code(responses[0].error);
             check_code(responses[1].error);
@@ -367,7 +376,7 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
             zkutil::addCheckNotExistsRequest(requests, *zk_client, processed_node_path);
 
         Coordination::Responses responses;
-        const auto code = zk_client->tryMulti(requests, responses);
+        code = zk_client->tryMulti(requests, responses);
         auto has_request_failed = [&](size_t request_index) { return responses[request_index]->error != Coordination::Error::ZOK; };
 
         if (code == Coordination::Error::ZOK)
@@ -401,6 +410,11 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
         /// most likely the processing node id path node was removed or created so let's try again
         LOG_TRACE(log, "Retrying setProcessing because processing node id path is unexpectedly missing or was created (error code: {})", code);
     }
+
+    throw Exception(
+        ErrorCodes::LOGICAL_ERROR,
+        "Failed to set file processing within {} retries, last error: {}",
+        max_num_tries, code);
 }
 
 void ObjectStorageQueueOrderedFileMetadata::prepareProcessedAtStartRequests(
diff --git a/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.cpp b/src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.cpp
@@ -512,7 +512,7 @@ ObjectStorageQueueSource::FileIterator::getNextKeyFromAcquiredBucket(size_t proc
         log, "Current processor: {}, acquired bucket: {}",
         processor, current_bucket_holder ? toString(current_bucket_holder->getBucket()) : "None");
 
-    while (true)
+    while (!shutdown_called)
     {
         /// Each processing thread gets next path
         /// and checks if corresponding bucket is already acquired by someone.
@@ -710,6 +710,7 @@ ObjectStorageQueueSource::FileIterator::getNextKeyFromAcquiredBucket(size_t proc
         if (listed_keys_cache.empty())
             return {};
     }
+    return {};
 }
 
 ObjectStorageQueueSource::ObjectStorageQueueSource(
diff --git a/src/Storages/ObjectStorageQueue/ObjectStorageQueueUnorderedFileMetadata.cpp b/src/Storages/ObjectStorageQueue/ObjectStorageQueueUnorderedFileMetadata.cpp
@@ -83,13 +83,15 @@ ObjectStorageQueueUnorderedFileMetadata::prepareProcessingRequestsImpl(Coordinat
 std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorageQueueUnorderedFileMetadata::setProcessingImpl()
 {
     const auto zk_client = getZooKeeper();
-    while (true)
+    const size_t max_num_tries = 1000;
+    Coordination::Error code;
+    for (size_t i = 0; i < max_num_tries; ++i)
     {
         Coordination::Requests requests;
         auto result_indexes = prepareProcessingRequestsImpl(requests);
 
         Coordination::Responses responses;
-        const auto code = zk_client->tryMulti(requests, responses);
+        code = zk_client->tryMulti(requests, responses);
         auto has_request_failed = [&](size_t request_index) { return responses[request_index]->error != Coordination::Error::ZOK; };
 
         if (code == Coordination::Error::ZOK)
@@ -118,6 +120,11 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
             log, "Retrying setProcessing because processing node id path "
             "is unexpectedly missing or was created (error code: {})", code);
     }
+
+    throw Exception(
+        ErrorCodes::LOGICAL_ERROR,
+        "Failed to set file processing within {} retries, last error: {}",
+        max_num_tries, code);
 }
 
 void ObjectStorageQueueUnorderedFileMetadata::prepareProcessedAtStartRequests(

Original file line number	Diff line number	Diff line change
`@@ -512,7 +512,7 @@ ObjectStorageQueueSource::FileIterator::getNextKeyFromAcquiredBucket(size_t proc`
`512`	`512`	`log, "Current processor: {}, acquired bucket: {}",`
`513`	`513`	`processor, current_bucket_holder ? toString(current_bucket_holder->getBucket()) : "None");`
`514`	`514`
`515`		`- while (true)`
	`515`	`+ while (!shutdown_called)`
`516`	`516`	`{`
`517`	`517`	`/// Each processing thread gets next path`
`518`	`518`	`/// and checks if corresponding bucket is already acquired by someone.`
`@@ -710,6 +710,7 @@ ObjectStorageQueueSource::FileIterator::getNextKeyFromAcquiredBucket(size_t proc`
`710`	`710`	`if (listed_keys_cache.empty())`
`711`	`711`	`return {};`
`712`	`712`	`}`
	`713`	`+ return {};`
`713`	`714`	`}`
`714`	`715`
`715`	`716`	`ObjectStorageQueueSource::ObjectStorageQueueSource(`