Fixed leader election in completely different way to allow many replicas in...

Fixed leader election in completely different way to allow many replicas in single ClickHouse instance (that is used in tests) [#METR-20132].

dbms/include/DB/Storages/MergeTree/AbandonableLockInZooKeeper.h

@@ -83,7 +83,7 @@ public:
 
 		try
 		{
-			zookeeper.tryRemove(holder_path);
+			zookeeper.tryRemoveEphemeralNodeWithRetries(holder_path);
 			zookeeper.trySet(path, ""); /// Это не обязательно.
 		}
 		catch (...)

libs/libzkutil/include/zkutil/LeaderElection.h

@@ -26,29 +26,6 @@ public:
 		std::string node_path = node->getPath();
 		node_name = node_path.substr(node_path.find_last_of('/') + 1);
 
-		/** Если есть ноды с таким же ephemeralOwner, то удалим их.
-		  * Такие ноды могли остаться после неуспешного удаления, если сессия при этом не истекла.
-		  */
-		zkutil::Stat node_stat;
-		zookeeper.get(node_path, &node_stat);
-
-		Strings brothers = zookeeper.getChildren(path);
-		for (const auto & brother : brothers)
-		{
-			if (brother == node_name)
-				continue;
-
-			zkutil::Stat brother_stat;
-			std::string brother_path = path + "/" + brother;
-			zookeeper.get(brother_path, &brother_stat);
-
-			if (brother_stat.ephemeralOwner == node_stat.ephemeralOwner)
-			{
-				LOG_WARNING(log, "Found obsolete ephemeral node from same session, removing: " + brother_path);
-				zookeeper.tryRemoveWithRetries(brother_path);
-			}
-		}
-
 		thread = std::thread(&LeaderElection::threadFunction, this);
 	}
 

libs/libzkutil/include/zkutil/ZooKeeper.h

@@ -17,10 +17,10 @@ namespace zkutil
 const UInt32 DEFAULT_SESSION_TIMEOUT = 30000;
 const UInt32 MEDIUM_SESSION_TIMEOUT = 120000;
 const UInt32 BIG_SESSION_TIMEOUT = 600000;
-const UInt32 DEFAULT_RETRY_NUM = 3;
 
 struct WatchWithEvent;
 
+
 /** Сессия в ZooKeeper. Интерфейс существенно отличается от обычного API ZooKeeper.
   * Вместо callback-ов для watch-ей используются Poco::Event. Для указанного события вызывается set() только при первом вызове watch.
   * Методы на чтение при восстанавливаемых ошибках OperationTimeout, ConnectionLoss пытаются еще retry_num раз.
@@ -31,7 +31,7 @@ struct WatchWithEvent;
 class ZooKeeper
 {
 public:
-	typedef std::shared_ptr<ZooKeeper> Ptr;
+	using Ptr = std::shared_ptr<ZooKeeper>;
 
 	ZooKeeper(const std::string & hosts, int32_t session_timeout_ms = DEFAULT_SESSION_TIMEOUT);
 
@@ -57,13 +57,11 @@ public:
 	  */
 	Ptr startNewSession() const;
 
-	int state();
-
 	/** Возвращает true, если сессия навсегда завершена.
 	  * Это возможно только если соединение было установлено, потом разорвалось, потом снова восстановилось, но слишком поздно.
-	  * Это достаточно редкая ситуация.
 	  * С другой стороны, если, например, указан неправильный сервер или порт, попытки соединения будут продолжаться бесконечно,
 	  *  expired() будет возвращать false, и все вызовы будут выбрасывать исключение ConnectionLoss.
+	  * Также возвращает true, если выставлен флаг is_dirty - просьба побыстрее завершить сессию.
 	  */
 	bool expired();
 
@@ -82,10 +80,10 @@ public:
 	  *  - Такая нода уже есть.
 	  * При остальных ошибках бросает исключение.
 	  */
-	int32_t tryCreate(const std::string & path, const std::string & data, int32_t mode, std::string & pathCreated);
+	int32_t tryCreate(const std::string & path, const std::string & data, int32_t mode, std::string & path_created);
 	int32_t tryCreate(const std::string & path, const std::string & data, int32_t mode);
 	int32_t tryCreateWithRetries(const std::string & path, const std::string & data, int32_t mode,
-								 std::string & pathCreated, size_t * attempt = nullptr);
+								 std::string & path_created, size_t * attempt = nullptr);
 
 	/** создает Persistent ноду.
 	 *  Игнорирует, если нода уже создана.
@@ -115,6 +113,31 @@ public:
 	/// Если есть проблемы с сетью может сам удалить ноду и вернуть ZNONODE
 	int32_t tryRemoveWithRetries(const std::string & path, int32_t version = -1, size_t * attempt = nullptr);
 
+	/** То же самое, но также выставляет флаг is_dirty, если все попытки удалить были неуспешными.
+	  * Это делается, потому что сессия может ещё жить после всех попыток, даже если прошло больше session_timeout времени.
+	  * Поэтому не стоит рассчитывать, что эфемерная нода действительно будет удалена.
+	  * Но флаг is_dirty позволит побыстрее завершить сессию.
+
+			Ridiculously Long Delay to Expire
+			When disconnects do happen, the common case should be a very* quick
+			reconnect to another server, but an extended network outage may
+			introduce a long delay before a client can reconnect to the ZooKeep‐
+			er service. Some developers wonder why the ZooKeeper client li‐
+			brary doesn’t simply decide at some point (perhaps twice the session
+			timeout) that enough is enough and kill the session itself.
+			There are two answers to this. First, ZooKeeper leaves this kind of
+			policy decision up to the developer. Developers can easily implement
+			such a policy by closing the handle themselves. Second, when a Zoo‐
+			Keeper ensemble goes down, time freezes. Thus, when the ensemble is
+			brought back up, session timeouts are restarted. If processes using
+			ZooKeeper hang in there, they may find out that the long timeout was
+			due to an extended ensemble failure that has recovered and pick right
+			up where they left off without any additional startup delay.
+
+			ZooKeeper: Distributed Process Coordination p118
+	  */
+	int32_t tryRemoveEphemeralNodeWithRetries(const std::string & path, int32_t version = -1, size_t * attempt = nullptr);
+
 	bool exists(const std::string & path, Stat * stat = nullptr, EventPtr watch = nullptr);
 
 	std::string get(const std::string & path, Stat * stat = nullptr, EventPtr watch = nullptr);
@@ -329,7 +352,7 @@ private:
 	}
 
 	/// методы не бросают исключений, а возвращают коды ошибок
-	int32_t createImpl(const std::string & path, const std::string & data, int32_t mode, std::string & pathCreated);
+	int32_t createImpl(const std::string & path, const std::string & data, int32_t mode, std::string & path_created);
 	int32_t removeImpl(const std::string & path, int32_t version = -1);
 	int32_t getImpl(const std::string & path, std::string & res, Stat * stat = nullptr, EventPtr watch = nullptr);
 	int32_t setImpl(const std::string & path, const std::string & data,
@@ -350,11 +373,17 @@ private:
 	std::unordered_set<WatchWithEvent *> watch_store;
 
 	/// Количество попыток повторить операцию чтения при OperationTimeout, ConnectionLoss
-	size_t retry_num = 3;
+	static constexpr size_t retry_num = 3;
 	Logger * log = nullptr;
+
+	/** При работе с сессией были неудачные попытки удалить эфемерные ноды,
+	  *  после которых лучше завершить сессию (чтобы эфемерные ноды всё-таки удалились)
+	  *  вместо того, чтобы продолжить пользоваться восстановившейся сессией.
+	  */
+	bool is_dirty = false;
 };
 
-typedef ZooKeeper::Ptr ZooKeeperPtr;
+using ZooKeeperPtr = ZooKeeper::Ptr;
 
 
 /** В конструкторе создает эфемерную ноду, в деструкторе - удаляет.
@@ -362,7 +391,7 @@ typedef ZooKeeper::Ptr ZooKeeperPtr;
 class EphemeralNodeHolder
 {
 public:
-	typedef Poco::SharedPtr<EphemeralNodeHolder> Ptr;
+	using Ptr = std::shared_ptr<EphemeralNodeHolder>;
 
 	EphemeralNodeHolder(const std::string & path_, ZooKeeper & zookeeper_, bool create, bool sequential, const std::string & data)
 		: path(path_), zookeeper(zookeeper_)
@@ -378,17 +407,17 @@ public:
 
 	static Ptr create(const std::string & path, ZooKeeper & zookeeper, const std::string & data = "")
 	{
-		return new EphemeralNodeHolder(path, zookeeper, true, false, data);
+		return std::make_shared<EphemeralNodeHolder>(path, zookeeper, true, false, data);
 	}
 
 	static Ptr createSequential(const std::string & path, ZooKeeper & zookeeper, const std::string & data = "")
 	{
-		return new EphemeralNodeHolder(path, zookeeper, true, true, data);
+		return std::make_shared<EphemeralNodeHolder>(path, zookeeper, true, true, data);
 	}
 
 	static Ptr existing(const std::string & path, ZooKeeper & zookeeper)
 	{
-		return new EphemeralNodeHolder(path, zookeeper, false, false, "");
+		return std::make_shared<EphemeralNodeHolder>(path, zookeeper, false, false, "");
 	}
 
 	~EphemeralNodeHolder()
@@ -398,35 +427,12 @@ public:
 			/** Важно, что в случае недоступности ZooKeeper, делаются повторные попытки удалить ноду.
 			  * Иначе возможна ситуация, когда объект EphemeralNodeHolder уничтожен,
 			  *  но сессия восстановится в течние session timeout, и эфемерная нода в ZooKeeper останется ещё надолго.
-			  * Но см. ниже - на самом деле, такая ситуация всё-равно возможна.
 			  */
-			zookeeper.tryRemoveWithRetries(path);
+			zookeeper.tryRemoveEphemeralNodeWithRetries(path);
 		}
 		catch (const KeeperException & e)
 		{
 			LOG_ERROR(zookeeper.log, "~EphemeralNodeHolder(): " << e.displayText());
-
-			/** На самом деле, сессия может ещё жить после этой ошибки.
-			  * Поэтому не стоит рассчитывать, что эфемерная нода действительно будет удалена.
-
-			  Ridiculously Long Delay to Expire
-			  When disconnects do happen, the common case should be a very* quick
-			  reconnect to another server, but an extended network outage may
-			  introduce a long delay before a client can reconnect to the ZooKeep‐
-			  er service. Some developers wonder why the ZooKeeper client li‐
-			  brary doesn’t simply decide at some point (perhaps twice the session
-			  timeout) that enough is enough and kill the session itself.
-			  There are two answers to this. First, ZooKeeper leaves this kind of
-			  policy decision up to the developer. Developers can easily implement
-			  such a policy by closing the handle themselves. Second, when a Zoo‐
-			  Keeper ensemble goes down, time freezes. Thus, when the ensemble is
-			  brought back up, session timeouts are restarted. If processes using
-			  ZooKeeper hang in there, they may find out that the long timeout was
-			  due to an extended ensemble failure that has recovered and pick right
-			  up where they left off without any additional startup delay.
-
-			  ZooKeeper: Distributed Process Coordination p118
-			  */
 		}
 	}
 
@@ -435,6 +441,6 @@ private:
 	ZooKeeper & zookeeper;
 };
 
-typedef EphemeralNodeHolder::Ptr EphemeralNodeHolderPtr;
+using EphemeralNodeHolderPtr = EphemeralNodeHolder::Ptr;
 
 }

libs/libzkutil/src/Lock.cpp

@@ -55,7 +55,7 @@ void Lock::unlock()
 			if (tryCheck() == Status::LOCKED_BY_ME)
 			{
 				size_t attempt;
-				int32_t code = zookeeper->tryRemoveWithRetries(lock_path, -1, &attempt);
+				int32_t code = zookeeper->tryRemoveEphemeralNodeWithRetries(lock_path, -1, &attempt);
 				if (attempt)
 				{
 					if (code != ZOK)
@@ -84,7 +84,7 @@ void Lock::unlock()
 Lock::Status Lock::tryCheck() const
 {
 	auto zookeeper = zookeeper_holder->getZooKeeper();
-	
+
 	Status lock_status;
 	Stat stat;
 	std::string dummy;

libs/libzkutil/src/RWLock.cpp

@@ -102,7 +102,7 @@ void RWLock::release()
 	if (key.empty())
 		throw DB::Exception{"RWLock: no lock is held", DB::ErrorCodes::LOGICAL_ERROR};
 
-	get_zookeeper()->remove(path + "/" + key);
+	get_zookeeper()->tryRemoveEphemeralNodeWithRetries(path + "/" + key);
 	key.clear();
 	owns_lock = false;
 }
@@ -189,7 +189,7 @@ void RWLock::acquireImpl(Mode mode)
 
 			if (mode == NonBlocking)
 			{
-				int32_t code = zookeeper->tryRemove(path + "/" + key);
+				int32_t code = zookeeper->tryRemoveEphemeralNodeWithRetries(path + "/" + key);
 				if (code == ZNONODE)
 					throw DB::Exception{"No such lock", DB::ErrorCodes::RWLOCK_NO_SUCH_LOCK};
 				else if (code != ZOK)
@@ -217,7 +217,7 @@ void RWLock::acquireImpl(Mode mode)
 		try
 		{
 			if (!key.empty())
-				get_zookeeper()->remove(path + "/" + key);
+				get_zookeeper()->tryRemoveEphemeralNodeWithRetries(path + "/" + key);
 		}
 		catch (...)
 		{

libs/libzkutil/src/SingleBarrier.cpp

@@ -128,7 +128,7 @@ void SingleBarrier::enter(uint64_t timeout)
 						break;
 					}
 
-					(void) zookeeper->tryRemove(path + "/tokens/" + cur_token);
+					(void) zookeeper->tryRemoveEphemeralNodeWithRetries(path + "/tokens/" + cur_token);
 					--token_count;
 				}
 			}
@@ -186,7 +186,7 @@ void SingleBarrier::abortIfRequested()
 			{
 				/// We have received a cancellation request while trying
 				/// to cross the barrier. Therefore we delete our token.
-				(void) get_zookeeper()->tryRemove(path + "/tokens/" + token);
+				(void) get_zookeeper()->tryRemoveEphemeralNodeWithRetries(path + "/tokens/" + token);
 			}
 			catch (...)
 			{

libs/libzkutil/src/ZooKeeper.cpp

@@ -197,7 +197,7 @@ int32_t ZooKeeper::tryGetChildren(const std::string & path, Strings & res,
 	return code;
 }
 
-int32_t ZooKeeper::createImpl(const std::string & path, const std::string & data, int32_t mode, std::string & pathCreated)
+int32_t ZooKeeper::createImpl(const std::string & path, const std::string & data, int32_t mode, std::string & path_created)
 {
 	int code;
 	/// имя ноды может быть больше переданного пути, если создается sequential нода.
@@ -210,7 +210,7 @@ int32_t ZooKeeper::createImpl(const std::string & path, const std::string & data
 
 	if (code == ZOK)
 	{
-		pathCreated = std::string(name_buffer);
+		path_created = std::string(name_buffer);
 	}
 
 	delete[] name_buffer;
@@ -220,14 +220,14 @@ int32_t ZooKeeper::createImpl(const std::string & path, const std::string & data
 
 std::string ZooKeeper::create(const std::string & path, const std::string & data, int32_t type)
 {
-	std::string  pathCreated;
-	check(tryCreate(path, data, type, pathCreated), path);
-	return pathCreated;
+	std::string path_created;
+	check(tryCreate(path, data, type, path_created), path);
+	return path_created;
 }
 
-int32_t ZooKeeper::tryCreate(const std::string & path, const std::string & data, int32_t mode, std::string & pathCreated)
+int32_t ZooKeeper::tryCreate(const std::string & path, const std::string & data, int32_t mode, std::string & path_created)
 {
-	int code = createImpl(path, data, mode, pathCreated);
+	int code = createImpl(path, data, mode, path_created);
 
 	if (!(	code == ZOK ||
 			code == ZNONODE ||
@@ -240,20 +240,20 @@ int32_t ZooKeeper::tryCreate(const std::string & path, const std::string & data,
 
 int32_t ZooKeeper::tryCreate(const std::string & path, const std::string & data, int32_t mode)
 {
-	std::string pathCreated;
-	return tryCreate(path, data, mode, pathCreated);
+	std::string path_created;
+	return tryCreate(path, data, mode, path_created);
 }
 
-int32_t ZooKeeper::tryCreateWithRetries(const std::string & path, const std::string & data, int32_t mode, std::string & pathCreated, size_t* attempt)
+int32_t ZooKeeper::tryCreateWithRetries(const std::string & path, const std::string & data, int32_t mode, std::string & path_created, size_t* attempt)
 {
-	return retry([&path, &data, mode, &pathCreated, this] { return tryCreate(path, data, mode, pathCreated); }, attempt);
+	return retry([&path, &data, mode, &path_created, this] { return tryCreate(path, data, mode, path_created); }, attempt);
 }
 
 
 void ZooKeeper::createIfNotExists(const std::string & path, const std::string & data)
 {
-	std::string pathCreated;
-	int32_t code = retry(std::bind(&ZooKeeper::createImpl, this, std::ref(path), std::ref(data), zkutil::CreateMode::Persistent, std::ref(pathCreated)));
+	std::string path_created;
+	int32_t code = retry(std::bind(&ZooKeeper::createImpl, this, std::ref(path), std::ref(data), zkutil::CreateMode::Persistent, std::ref(path_created)));
 
 	if (code == ZOK || code == ZNODEEXISTS)
 		return;
@@ -314,10 +314,29 @@ int32_t ZooKeeper::tryRemoveWithRetries(const std::string & path, int32_t versio
 			code == ZNONODE ||
 			code == ZBADVERSION ||
 			code == ZNOTEMPTY))
+	{
 		throw KeeperException(code, path);
+	}
+
 	return code;
 }
 
+int32_t ZooKeeper::tryRemoveEphemeralNodeWithRetries(const std::string & path, int32_t version, size_t * attempt)
+{
+	try
+	{
+		return tryRemoveWithRetries(path, version, attempt);
+	}
+	catch (const KeeperException &)
+	{
+		// Установим флажок, который говорит о том, что сессию лучше воспринимать так же как истёкшую,
+		///  чтобы кто-нибудь её пересоздал, и, в случае эфемерной ноды, нода всё-таки была удалена.
+		is_dirty = true;
+
+		throw;
+	}
+}
+
 int32_t ZooKeeper::existsImpl(const std::string & path, Stat * stat_, EventPtr watch)
 {
 	int32_t code;
@@ -625,14 +644,9 @@ std::string ZooKeeper::error2string(int32_t code)
 	return zerror(code);
 }
 
-int ZooKeeper::state()
-{
-	return zoo_state(impl);
-}
-
 bool ZooKeeper::expired()
 {
-	return state() == ZOO_EXPIRED_SESSION_STATE;
+	return is_dirty || zoo_state(impl) == ZOO_EXPIRED_SESSION_STATE;
 }
 
 int64_t ZooKeeper::getClientID()