ClickHouse · devcrafter · Nov 10, 2022 · Oct 24, 2022 · Oct 24, 2022 · Oct 26, 2022
diff --git a/src/Common/ZooKeeper/TestKeeper.h b/src/Common/ZooKeeper/TestKeeper.h
@@ -34,7 +34,7 @@ using TestKeeperRequestPtr = std::shared_ptr<TestKeeperRequest>;
 class TestKeeper final : public IKeeper
 {
 public:
-    TestKeeper(const zkutil::ZooKeeperArgs & args_);
+    explicit TestKeeper(const zkutil::ZooKeeperArgs & args_);
     ~TestKeeper() override;
 
     bool isExpired() const override { return expired; }

diff --git a/src/Common/ZooKeeper/ZooKeeper.h b/src/Common/ZooKeeper/ZooKeeper.h
@@ -156,7 +156,7 @@ class ZooKeeper
     using Ptr = std::shared_ptr<ZooKeeper>;
     using ErrorsList = std::initializer_list<Coordination::Error>;
 
-    ZooKeeper(const ZooKeeperArgs & args_, std::shared_ptr<DB::ZooKeeperLog> zk_log_ = nullptr);
+    explicit ZooKeeper(const ZooKeeperArgs & args_, std::shared_ptr<DB::ZooKeeperLog> zk_log_ = nullptr);
 
     /** Config of the form:
         <zookeeper>

diff --git a/src/Core/Settings.h b/src/Core/Settings.h
@@ -658,6 +658,11 @@ static constexpr UInt64 operator""_GiB(unsigned long long value)
     M(Bool, multiple_joins_try_to_keep_original_names, false, "Do not add aliases to top level expression list on multiple joins rewrite", 0) \
     M(Bool, optimize_distinct_in_order, true, "Enable DISTINCT optimization if some columns in DISTINCT form a prefix of sorting. For example, prefix of sorting key in merge tree or ORDER BY statement", 0) \
     M(Bool, optimize_sorting_by_input_stream_properties, true, "Optimize sorting by sorting properties of input stream", 0) \
+    M(UInt64, insert_keeper_max_retries, 0, "Max retries for keeper operations during insert", 0) \
+    M(UInt64, insert_keeper_retry_initial_backoff_ms, 100, "Initial backoff timeout for keeper operations during insert", 0) \
+    M(UInt64, insert_keeper_retry_max_backoff_ms, 10000, "Max backoff timeout for keeper operations during insert", 0) \
+    M(Float, insert_keeper_fault_injection_probability, 0.0f, "Approximate probability of failure for a keeper request during insert. Valid value is in interval [0.0f, 1.0f]", 0) \
+    M(UInt64, insert_keeper_fault_injection_seed, 0, "0 - random seed, otherwise the setting value", 0) \
     // End of COMMON_SETTINGS
     // Please add settings related to formats into the FORMAT_FACTORY_SETTINGS and move obsolete settings to OBSOLETE_SETTINGS.
 

diff --git a/src/Storages/MergeTree/EphemeralLockInZooKeeper.cpp b/src/Storages/MergeTree/EphemeralLockInZooKeeper.cpp
@@ -2,6 +2,7 @@
 #include <Common/ZooKeeper/KeeperException.h>
 #include <Common/logger_useful.h>
 #include <base/types.h>
+#include <Storages/MergeTree/ZooKeeperWithFaultInjection.h>
 
 
 namespace DB
@@ -12,22 +13,22 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
 }
 
-EphemeralLockInZooKeeper::EphemeralLockInZooKeeper(const String & path_prefix_, zkutil::ZooKeeper & zookeeper_, const String & path_)
-    : zookeeper(&zookeeper_), path_prefix(path_prefix_), path(path_)
+EphemeralLockInZooKeeper::EphemeralLockInZooKeeper(const String & path_prefix_, const ZooKeeperWithFaultInjectionPtr & zookeeper_, const String & path_)
+    : zookeeper(zookeeper_), path_prefix(path_prefix_), path(path_)
 {
     if (path.size() <= path_prefix.size())
         throw Exception("Logical error: name of the main node is shorter than prefix.", ErrorCodes::LOGICAL_ERROR);
 }
 
 std::optional<EphemeralLockInZooKeeper> createEphemeralLockInZooKeeper(
-    const String & path_prefix_, const String & temp_path, zkutil::ZooKeeper & zookeeper_, const String & deduplication_path)
+    const String & path_prefix_, const String & temp_path, const ZooKeeperWithFaultInjectionPtr & zookeeper_, const String & deduplication_path)
 {
     String path;
 
     if (deduplication_path.empty())
     {
         String holder_path = temp_path + "/" + EphemeralLockInZooKeeper::LEGACY_LOCK_OTHER;
-        path = zookeeper_.create(path_prefix_, holder_path, zkutil::CreateMode::EphemeralSequential);
+        path = zookeeper_->create(path_prefix_, holder_path, zkutil::CreateMode::EphemeralSequential);
     }
     else
     {
@@ -39,11 +40,15 @@ std::optional<EphemeralLockInZooKeeper> createEphemeralLockInZooKeeper(
         ops.emplace_back(zkutil::makeRemoveRequest(deduplication_path, -1));
         ops.emplace_back(zkutil::makeCreateRequest(path_prefix_, holder_path, zkutil::CreateMode::EphemeralSequential));
         Coordination::Responses responses;
-        Coordination::Error e = zookeeper_.tryMulti(ops, responses);
+        Coordination::Error e = zookeeper_->tryMulti(ops, responses);
         if (e != Coordination::Error::ZOK)
         {
             if (responses[0]->error == Coordination::Error::ZNODEEXISTS)
             {
+                LOG_DEBUG(
+                    &Poco::Logger::get("createEphemeralLockInZooKeeper"),
+                    "Deduplication path already exists: deduplication_path={}",
+                    deduplication_path);
                 return {};
             }
             else
@@ -82,9 +87,31 @@ EphemeralLockInZooKeeper::~EphemeralLockInZooKeeper()
     {
         unlock();
     }
+    catch (const zkutil::KeeperException & e)
+    {
+        if (Coordination::isHardwareError(e.code))
+            LOG_DEBUG(
+                &Poco::Logger::get("EphemeralLockInZooKeeper"),
+                "ZooKeeper communication error during unlock: code={} message='{}'",
+                e.code,
+                e.message());
+        else if (e.code == Coordination::Error::ZNONODE)
+            /// To avoid additional round-trip for unlocking,
+            /// ephemeral node can be deleted explicitly as part of another multi op request to ZK
+            /// and marked as such via assumeUnlocked() if we got successful response.
+            /// But it's possible that the multi op request can be executed on server side, and client will not get response due to network issue.
+            /// In such case, assumeUnlocked() will not be called, so we'll get ZNONODE error here since the noded is already deleted
+            LOG_DEBUG(
+                &Poco::Logger::get("EphemeralLockInZooKeeper"),
+                "ZooKeeper node was already deleted: code={} message={}",
+                e.code,
+                e.message());
+        else
+            tryLogCurrentException("EphemeralLockInZooKeeper");
+    }
     catch (...)
     {
-        tryLogCurrentException("~EphemeralLockInZooKeeper");
+        tryLogCurrentException("EphemeralLockInZooKeeper");
     }
 }
 

diff --git a/src/Storages/MergeTree/EphemeralLockInZooKeeper.h b/src/Storages/MergeTree/EphemeralLockInZooKeeper.h
@@ -12,6 +12,8 @@
 
 namespace DB
 {
+class ZooKeeperWithFaultInjection;
+using ZooKeeperWithFaultInjectionPtr = std::shared_ptr<ZooKeeperWithFaultInjection>;
 
 namespace ErrorCodes
 {
@@ -25,13 +27,14 @@ namespace ErrorCodes
 class EphemeralLockInZooKeeper : public boost::noncopyable
 {
     friend std::optional<EphemeralLockInZooKeeper> createEphemeralLockInZooKeeper(
-        const String & path_prefix_, const String & temp_path, zkutil::ZooKeeper & zookeeper_, const String & deduplication_path);
+        const String & path_prefix_, const String & temp_path, const ZooKeeperWithFaultInjectionPtr & zookeeper_, const String & deduplication_path);
 
 protected:
-    EphemeralLockInZooKeeper() = delete;
-    EphemeralLockInZooKeeper(const String & path_prefix_, zkutil::ZooKeeper & zookeeper_, const String & path_);
+    EphemeralLockInZooKeeper(const String & path_prefix_, const ZooKeeperWithFaultInjectionPtr & zookeeper_, const String & path_);
 
 public:
+    EphemeralLockInZooKeeper() = delete;
+
     /// Fake "secondary node" names for blocks with and without "deduplication_path"
     static constexpr const char * LEGACY_LOCK_INSERT = "abandonable_lock-insert";
     static constexpr const char * LEGACY_LOCK_OTHER = "abandonable_lock-other";
@@ -53,7 +56,7 @@ class EphemeralLockInZooKeeper : public boost::noncopyable
 
     bool isLocked() const
     {
-        return zookeeper;
+        return zookeeper.get();
     }
 
     String getPath() const
@@ -91,13 +94,13 @@ class EphemeralLockInZooKeeper : public boost::noncopyable
     ~EphemeralLockInZooKeeper();
 
 private:
-    zkutil::ZooKeeper * zookeeper = nullptr;
+    ZooKeeperWithFaultInjectionPtr zookeeper;
     String path_prefix;
     String path;
 };
 
 std::optional<EphemeralLockInZooKeeper> createEphemeralLockInZooKeeper(
-    const String & path_prefix_, const String & temp_path, zkutil::ZooKeeper & zookeeper_, const String & deduplication_path);
+    const String & path_prefix_, const String & temp_path, const ZooKeeperWithFaultInjectionPtr & zookeeper_, const String & deduplication_path);
 
 
 /// Acquires block number locks in all partitions.

diff --git a/src/Storages/MergeTree/ReplicatedMergeTreeMergeStrategyPicker.cpp b/src/Storages/MergeTree/ReplicatedMergeTreeMergeStrategyPicker.cpp
@@ -112,6 +112,8 @@ void ReplicatedMergeTreeMergeStrategyPicker::refreshState()
         && now - last_refresh_time < REFRESH_STATE_MINIMUM_INTERVAL_SECONDS)
         return;
 
+    LOG_DEBUG(storage.log, "Updating strategy picker state");
+
     auto zookeeper = storage.getZooKeeper();
     auto all_replicas = zookeeper->getChildren(storage.zookeeper_path + "/replicas");
 
@@ -154,6 +156,8 @@ void ReplicatedMergeTreeMergeStrategyPicker::refreshState()
     last_refresh_time = now;
     current_replica_index = current_replica_index_tmp;
     active_replicas = active_replicas_tmp;
+
+    LOG_DEBUG(storage.log, "Strategy picker state updated, current replica: {}, active replicas: [{}]", current_replica_index, fmt::join(active_replicas, ", "));
 }