From 0b041a2af8bdc0340d9d41b301b1ccc973056800 Mon Sep 17 00:00:00 2001
From: Yunze Xu <xyzinfernity@163.com>
Date: Tue, 29 Sep 2020 08:50:17 +0800
Subject: [PATCH] [pulsar-admin-tools] Support delete all data associated with
 a cluster (#8133)

### Motivation

When multiple broker clusters shared the same bookie cluster, if user wanted to remove a broker cluster, the associated ledgers in bookies should also be deleted.

### Modifications

- Add an option to `cluster delete` command to delete all the data associated with the cluster.

Currently there's no way to delete the ledgers of schema from client's side, a new REST API to delete schema's ledgers should be exposed or the existed `SchemaRegistryServiceImpl#deleteSchema`'s semantics should be change.

### Verifying this change
The new option has been tested in local environment by:
1. Deploy a ZK which acts as both local ZK and the configuration store;
2. Init metadata of 2 clusters, the 2nd cluster use `--bookkeeper-metadata-service` option to share the same BK metadata position;
3. Deploy a BK;
4. Deploy 2 brokers associated with the 2 clusters' metadata;
5. Create some topics and produce some messages to 2 clusters;
6. Run `bin/pulsar cluster -all <cluster>` to delete the whole cluster, use `bin/bookkeeper shell list ledgers` to verify the change of ledgers count.

**An integration test is needed to simulate the procedures above.**
---
 .../apache/pulsar/admin/cli/CmdClusters.java  | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
index 8ea2e36ad5c66..a4893985606d7 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
@@ -119,8 +119,27 @@ private class Delete extends CliCommand {
         @Parameter(description = "cluster-name\n", required = true)
         private java.util.List<String> params;
 
+        @Parameter(names = { "-a", "--all" }, description = "Delete all data (tenants) of the cluster\n", required = false)
+        private boolean deleteAll = false;
+
         void run() throws PulsarAdminException {
             String cluster = getOneArgument(params);
+
+            if (deleteAll) {
+                for (String tenant : admin.tenants().getTenants()) {
+                    for (String namespace : admin.namespaces().getNamespaces(tenant)) {
+                        for (String topic : admin.topics().getList(namespace)) {
+                            admin.topics().delete(topic, true);
+                            // TODO: Delete all the ledgers of the SchemaStorage
+                            // admin.schemas().deleteSchema(topic) won't delete the schema's ledger. Instead a new ledger will be created.
+                            //       https://github.com/apache/pulsar/issues/8134
+                        }
+                        admin.namespaces().deleteNamespace(namespace);
+                    }
+                    admin.tenants().deleteTenant(tenant);
+                }
+            }
+
             admin.clusters().deleteCluster(cluster);
         }
     }