Fix lock_object_storage_task_distribution_ms with lost host

ianton-ru · ianton-ru · commit 45acc4d47b7f · 2025-09-29T16:44:57.000+02:00
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp
@@ -186,7 +186,7 @@ std::optional<String> StorageObjectStorageStableTaskDistributor::getMatchingFile
         // Queue file for its assigned replica
         {
             std::lock_guard lock(mutex);
-            unprocessed_files[file_path] = number_of_current_replica;
+            unprocessed_files[file_path] = file_replica_idx;
             connection_to_files[file_replica_idx].push_back(file_path);
         }
     }
@@ -277,7 +277,11 @@ void StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(size_
 
     replica_to_files_to_be_processed.erase(number_of_current_replica);
     for (const auto & file_path : processed_file_list_ptr->second)
-        unprocessed_files[file_path] = getReplicaForFile(file_path);
+    {
+        auto file_replica_idx = getReplicaForFile(file_path);
+        unprocessed_files[file_path] = file_replica_idx;
+        connection_to_files[file_replica_idx].push_back(file_path);
+    }
 }
 
 }
diff --git a/tests/integration/test_s3_cache_locality/test.py b/tests/integration/test_s3_cache_locality/test.py
@@ -86,8 +86,7 @@ def check_s3_gets(cluster, node, expected_result, cluster_first, cluster_second,
         "filesystem_cache_name": "'raw_s3_cache'",
         }
 
-    if lock_object_storage_task_distribution_ms > 0:
-        settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms
+    settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms
 
     query_id_first = str(uuid.uuid4())
     result_first = node.query(

Original file line number	Diff line number	Diff line change
`@@ -186,7 +186,7 @@ std::optional<String> StorageObjectStorageStableTaskDistributor::getMatchingFile`
`186`	`186`	`// Queue file for its assigned replica`
`187`	`187`	`{`
`188`	`188`	`std::lock_guard lock(mutex);`
`189`		`- unprocessed_files[file_path] = number_of_current_replica;`
	`189`	`+ unprocessed_files[file_path] = file_replica_idx;`
`190`	`190`	`connection_to_files[file_replica_idx].push_back(file_path);`
`191`	`191`	`}`
`192`	`192`	`}`
`@@ -277,7 +277,11 @@ void StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(size_`
`277`	`277`
`278`	`278`	`replica_to_files_to_be_processed.erase(number_of_current_replica);`
`279`	`279`	`for (const auto & file_path : processed_file_list_ptr->second)`
`280`		`- unprocessed_files[file_path] = getReplicaForFile(file_path);`
	`280`	`+ {`
	`281`	`+ auto file_replica_idx = getReplicaForFile(file_path);`
	`282`	`+ unprocessed_files[file_path] = file_replica_idx;`
	`283`	`+ connection_to_files[file_replica_idx].push_back(file_path);`
	`284`	`+ }`
`281`	`285`	`}`
`282`	`286`
`283`	`287`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,8 +86,7 @@ def check_s3_gets(cluster, node, expected_result, cluster_first, cluster_second,`
`86`	`86`	`"filesystem_cache_name": "'raw_s3_cache'",`
`87`	`87`	`}`
`88`	`88`
`89`		`- if lock_object_storage_task_distribution_ms > 0:`
`90`		`- settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms`
	`89`	`+ settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms`
`91`	`90`
`92`	`91`	`query_id_first = str(uuid.uuid4())`
`93`	`92`	`result_first = node.query(`