From 3b6ef8b28c609556c047c5b3b5665f5cf05c18b5 Mon Sep 17 00:00:00 2001
From: Alexey Milovidov <milovidov@yandex-team.ru>
Date: Fri, 11 Sep 2015 00:32:33 +0300
Subject: [PATCH] dbms: quorum inserts: development [#METR-16779].

---
 dbms/include/DB/Core/ErrorCodes.h             |  1 +
 .../ReplicatedMergeTreeBlockOutputStream.h    | 29 ++++++++++++++++++-
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/dbms/include/DB/Core/ErrorCodes.h b/dbms/include/DB/Core/ErrorCodes.h
index 1262cf3163c..bdfd2893352 100644
--- a/dbms/include/DB/Core/ErrorCodes.h
+++ b/dbms/include/DB/Core/ErrorCodes.h
@@ -287,6 +287,7 @@ namespace ErrorCodes
 		DICTIONARY_IS_EMPTY = 281,
 		INCORRECT_INDEX = 282,
 		UNKNOWN_GLOBAL_SUBQUERIES_METHOD = 284,
+		TOO_LESS_LIVE_REPLICAS = 285,
 
 		KEEPER_EXCEPTION = 999,
 		POCO_EXCEPTION = 1000,
diff --git a/dbms/include/DB/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h b/dbms/include/DB/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h
index abf2b71af41..ae89ff016b3 100644
--- a/dbms/include/DB/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h
+++ b/dbms/include/DB/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h
@@ -3,6 +3,7 @@
 #include <DB/Storages/StorageReplicatedMergeTree.h>
 #include <DB/Storages/MergeTree/AbandonableLockInZooKeeper.h>
 #include <DB/DataStreams/IBlockOutputStream.h>
+#include <DB/IO/Operators.h>
 
 
 namespace DB
@@ -32,7 +33,33 @@ public:
 		  */
 		if (quorum)
 		{
-			// TODO
+			/// Список живых реплик. Все они регистрируют эфемерную ноду для leader_election.
+			auto live_replicas = zookeeper->getChildren(storage.zookeeper_path + "/leader_election");
+
+			if (live_replicas.size() < quorum)
+			{
+				String list_of_replicas;
+
+				if (live_replicas.empty())
+					list_of_replicas = "none";
+				else
+				{
+					WriteBufferFromString out(list_of_replicas);
+					for (auto it = live_replicas.begin(); it != live_replicas.end(); ++it)
+						out << (it == live_replicas.begin() ? "" : ", ") << *it;
+				}
+
+				throw Exception("Number of alive replicas ("
+					+ toString(live_replicas.size()) + ") is less than requested quorum ("
+					+ toString(quorum) + "). Alive replicas: " + list_of_replicas,
+					ErrorCodes::TOO_LESS_LIVE_REPLICAS);
+			}
+
+			/// Разумеется, реплики могут перестать быть живыми после этой проверки. Это не проблема.
+
+			/** Есть ли у нас последний кусок, записанный с кворумом?
+			  * В ZK будем иметь следующую структуру директорий:
+			  */
 		}
 
 		auto part_blocks = storage.writer.splitBlockIntoParts(block);