Merge pull request ClickHouse#86874 from ClickHouse/backport/25.8/86615

Backport ClickHouse#86615 to 25.8: s3queue: fix logical error because of keeper session expired

Author: clickhouse-gh[bot]

src/Storages/ObjectStorageQueue/ObjectStorageQueueOrderedFileMetadata.cpp

@@ -278,7 +278,7 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
     processing_id = node_metadata.processing_id = getRandomASCIIString(10);
     auto processor_info = getProcessorInfo(processing_id.value());
 
-    const size_t max_num_tries = 1000;
+    const size_t max_num_tries = 100;
     Coordination::Error code;
     for (size_t i = 0; i < max_num_tries; ++i)
     {
@@ -428,7 +428,7 @@ std::pair<bool, ObjectStorageQueueIFileMetadata::FileStatus::State> ObjectStorag
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state of zookeeper transaction: {}", code);
 
         /// 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);
+        LOG_DEBUG(log, "Retrying setProcessing because processing node id path is unexpectedly missing or was created (error code: {})", code);
     }
 
     throw Exception(

src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.cpp

@@ -593,6 +593,18 @@ ObjectStorageQueueSource::FileIterator::getNextKeyFromAcquiredBucket(size_t proc
         if (current_bucket_holder && current_bucket_holder->isZooKeeperSessionExpired())
         {
             LOG_TRACE(log, "ZooKeeper session expired, bucket {} not longer hold", current_bucket_holder->getBucket());
+
+            for (auto & [bucket, bucket_info] : listed_keys_cache)
+            {
+                /// Reset current processor for the keys
+                /// to avoid the above error "Expected current processor {} to be equal to {} for bucket {}".
+                if (bucket_info.processor.has_value() && bucket_info.processor.value() == current_processor)
+                {
+                    LOG_DEBUG(log, "Resetting processor ({}) for bucket {}", current_processor, bucket);
+                    bucket_info.processor.reset();
+                }
+            }
+
             current_bucket_holder = {};
         }