Attempt to not interrupt merges when ZK session is reinitialized [#METR-22283].

dbms/src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.cpp

@@ -171,6 +171,10 @@ void ReplicatedMergeTreeRestartingThread::run()
 		storage.remote_part_checker_endpoint_holder->cancel();
 		storage.remote_part_checker_endpoint_holder = nullptr;
 
+		storage.merger.cancel();
+		if (storage.unreplicated_merger)
+			storage.unreplicated_merger->cancel();
+
 		partialShutdown();
 	}
 	catch (...)
@@ -211,10 +215,6 @@ bool ReplicatedMergeTreeRestartingThread::tryStartup()
 			std::bind(&StorageReplicatedMergeTree::queueTask, &storage, std::placeholders::_1));
 		storage.queue_task_handle->wake();
 
-		storage.merger.uncancel();
-		if (storage.unreplicated_merger)
-			storage.unreplicated_merger->uncancel();
-
 		return true;
 	}
 	catch (...)
@@ -362,10 +362,6 @@ void ReplicatedMergeTreeRestartingThread::partialShutdown()
 	storage.alter_query_event->set();
 	storage.replica_is_active_node = nullptr;
 
-	storage.merger.cancel();
-	if (storage.unreplicated_merger)
-		storage.unreplicated_merger->cancel();
-
 	LOG_TRACE(log, "Waiting for threads to finish");
 	if (storage.is_leader_node)
 	{

dbms/src/Storages/StorageReplicatedMergeTree.cpp

@@ -1162,7 +1162,7 @@ bool StorageReplicatedMergeTree::executeLogEntry(const LogEntry & entry, Backgro
 			if (!do_fetch)
 			{
 				merger.renameMergedTemporaryPart(parts, part, entry.new_part_name, &transaction);
-				zookeeper->multi(ops);
+				getZooKeeper()->multi(ops);		/// After long merge, get fresh ZK handle, because previous session may be expired.
 
 				/** Удаление старых кусков из ZK и с диска делается отложенно - см. ReplicatedMergeTreeCleanupThread, clearOldParts.
 				  */