Add test_ddl_worker_replicas

tests/integration/test_ddl_worker_replicas/configs/remote_servers.xml

@@ -0,0 +1,30 @@
+<clickhouse>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <host>node3</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node4</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+
+    <allow_zookeeper_write>1</allow_zookeeper_write>
+</clickhouse>

tests/integration/test_ddl_worker_replicas/test.py

@@ -0,0 +1,67 @@
+import pytest
+import time
+
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+
+node1 = cluster.add_instance(
+    "node1",
+    main_configs=["configs/remote_servers.xml"],
+    with_zookeeper=True,
+    stay_alive=True,
+)
+node2 = cluster.add_instance(
+    "node2", main_configs=["configs/remote_servers.xml"], with_zookeeper=True
+)
+node3 = cluster.add_instance(
+    "node3", main_configs=["configs/remote_servers.xml"], with_zookeeper=True
+)
+node4 = cluster.add_instance(
+    "node4", main_configs=["configs/remote_servers.xml"], with_zookeeper=True
+)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def test_ddl_worker_replicas(started_cluster):
+    replica_list = node1.query(
+        "SELECT name FROM system.zookeeper WHERE path='/clickhouse/task_queue/replicas'"
+    ).strip()
+
+    replica_list = list(replica_list.split("\n"))
+    expected_replicas = ["node1:9000", "node2:9000", "node3:9000", "node4:9000"]
+    assert expected_replicas.sort() == replica_list.sort()
+
+    for replica in replica_list:
+        result = node1.query(
+            f"SELECT name, value, ephemeralOwner FROM system.zookeeper WHERE path='/clickhouse/task_queue/replicas/{replica}'"
+        ).strip()
+
+        lines = list(result.split("\n"))
+        assert len(lines) == 1
+
+        parts = list(lines[0].split("\t"))
+        assert len(parts) == 3
+        assert parts[0] == "active"
+        assert len(parts[1]) != 0
+        assert len(parts[2]) != 0
+
+    node4.stop()
+    time.sleep(1)
+
+    result = node1.query(
+        f"SELECT name, value, ephemeralOwner FROM system.zookeeper WHERE path='/clickhouse/task_queue/replicas/node4:9000'"
+    ).strip()
+
+    lines = list(result.split("\n"))
+    assert len(lines) == 1
+    assert len(lines[0]) == 0