only drop inactive replica

update doc

Increase timeout to release the zookeeper Ephemeral nodes

Fix code comment

use PartitionManager

make integrations test passed

docs/en/sql-reference/statements/alter.md

@@ -198,12 +198,14 @@ Constraint check *will not be executed* on existing data if it was added.
 All changes on replicated tables are broadcasting to ZooKeeper so will be applied on other replicas.
 
 ### Manipulations with Replicas {#manipulations-with-replicas}
+
 Replicas can be dropped using following syntax:
+
 ```sql
 ALTER TABLE [db].name DROP REPLICA replica_name;
 ```
 
-Queries will remove the replica path in zookeeper, it's useful when you want to decrease your replica factor. If the replica_name is local replica, then it behavious same as `DROP TABLE`;
+Queries will remove the replica path in zookeeper, it's useful when you want to decrease your replica factor. It will only drop the inactive/stale replica, and it can't drop local replica, please use `DROP TABLE` for that.
 
 ### Manipulations with Partitions and Parts {#alter_manipulations-with-partitions}
 

src/Storages/StorageReplicatedMergeTree.cpp

@@ -884,6 +884,14 @@ void StorageReplicatedMergeTree::removeReplica(const String & replica)
         throw Exception("Table was not dropped because ZooKeeper session has expired.", ErrorCodes::TABLE_WAS_NOT_DROPPED);
 
     auto to_drop_path = zookeeper_path + "/replicas/" + replica;
+
+    //check if is active replica if we drop other replicas
+    if (replica != replica_name && zookeeper->exists(to_drop_path + "/is_active"))
+    {
+        throw Exception("Can't remove replica: " + replica + ", because it's active",
+            ErrorCodes::LOGICAL_ERROR);
+    }
+
     LOG_INFO(log, "Removing replica " << to_drop_path);
     /// It may left some garbage if to_drop_path subtree are concurently modified
     zookeeper->tryRemoveRecursive(to_drop_path);
@@ -4123,9 +4131,10 @@ void StorageReplicatedMergeTree::dropReplica(const String & replica)
 {
     if (replica_name == replica)
     {
-        drop();
-        return;
+        throw Exception("We can't drop local replica, please use `DROP TABLE` if you want to clean the data and drop this replica",
+            ErrorCodes::LOGICAL_ERROR);
     }
+    // remove other replicas
     removeReplica(replica);
 }
 

src/Storages/StorageReplicatedMergeTree.h

@@ -115,7 +115,7 @@ public:
       */
     void drop() override;
 
-    /** Removes a specific replica from Zookeeper. If replica is local, it works same as `drop` method.
+    /** Removes a specific replica from Zookeeper.
      */
     void dropReplica(const String & replica_name);
 

tests/integration/test_drop_replica/test.py

@@ -2,7 +2,9 @@ import time
 import pytest
 
 from helpers.cluster import ClickHouseCluster
+from helpers.cluster import ClickHouseKiller
 from helpers.test_tools import assert_eq_with_retry
+from helpers.network import PartitionManager
 
 def fill_nodes(nodes, shard):
     for node in nodes:
@@ -39,15 +41,19 @@ def test_drop_replica(start_cluster):
     for i in range(100):
         node_1_1.query("INSERT INTO test.test_table VALUES (1, {})".format(i))
 
-    node_1_2.kill_clickhouse()
-
     zk = cluster.get_kazoo_client('zoo1')
+
+    assert "can't drop local replica" in node_1_1.query_and_get_error("ALTER TABLE test.test_table drop replica 'node_1_1'")
+    assert "can't drop local replica" in node_1_2.query_and_get_error("ALTER TABLE test.test_table drop replica 'node_1_2'")
+    assert "it's active" in node_1_1.query_and_get_error("ALTER TABLE test.test_table drop replica 'node_1_2'")
+
+    with PartitionManager() as pm:
+        node_1_2.kill_clickhouse()
+        pm.drop_instance_zk_connections(node_1_2)
+        time.sleep(120)
         node_1_1.query("ALTER TABLE test.test_table drop replica 'node_1_2'")
-    e2 = zk.exists("/clickhouse/tables/test/{shard}/replicated/replicas/{replica}".format(shard=1, replica='node_1_2'))
-
-    assert(e2 == False)
-
-    node_1_1.query("ALTER TABLE test.test_table drop replica 'node_1_1'")
+        exists_replica_1_2 = zk.exists("/clickhouse/tables/test/{shard}/replicated/replicas/{replica}".format(shard=1, replica='node_1_2'))
+        assert (exists_replica_1_2 == None)
+        node_1_1.query("DROP TABLE test.test_table")
         exists_base_path = zk.exists("/clickhouse/tables/test/{shard}/replicated".format(shard=1))
-
-    assert(exists_base_path == False)
+        assert(exists_base_path == None)