From 6c876703c96dac693524845ce897e3b51f499ea4 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 30 Sep 2020 19:04:10 +0800 Subject: [PATCH] Add command to delete a cluster's metadata from ZK --- bin/pulsar | 3 + .../pulsar/PulsarClusterMetadataTeardown.java | 111 ++++++++++++++++++ 2 files changed, 114 insertions(+) create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataTeardown.java diff --git a/bin/pulsar b/bin/pulsar index 1fe53324b69f7..0c20c8d56bda3 100755 --- a/bin/pulsar +++ b/bin/pulsar @@ -143,6 +143,7 @@ where command is one of: standalone Run a broker server with local bookies and local zookeeper initialize-cluster-metadata One-time metadata initialization + delete-cluster-metadata Delete a cluster's metadata initialize-transaction-coordinator-metadata One-time transaction coordinator metadata initialization initialize-namespace namespace initialization compact-topic Run compaction against a topic @@ -339,6 +340,8 @@ elif [ $COMMAND == "standalone" ]; then exec $JAVA $OPTS $ASPECTJ_AGENT ${ZK_OPTS} -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@ elif [ $COMMAND == "initialize-cluster-metadata" ]; then exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@ +elif [ $COMMAND == "delete-cluster-metadata" ]; then + exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataTeardown $@ elif [ $COMMAND == "initialize-transaction-coordinator-metadata" ]; then exec $JAVA $OPTS org.apache.pulsar.PulsarTransactionCoordinatorMetadataSetup $@ elif [ $COMMAND == "initialize-namespace" ]; then diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataTeardown.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataTeardown.java new file mode 100644 index 0000000000000..a9a1c11b7df1c --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataTeardown.java @@ -0,0 +1,111 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; +import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZKUtil; +import org.apache.zookeeper.ZooKeeper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ExecutionException; + +/** + * Teardown the metadata for a existed Pulsar cluster + */ +public class PulsarClusterMetadataTeardown { + + private static class Arguments { + @Parameter(names = { "-zk", + "--zookeeper"}, description = "Local ZooKeeper quorum connection string", required = true) + private String zookeeper; + + @Parameter(names = { + "--zookeeper-session-timeout-ms" + }, description = "Local zookeeper session timeout ms") + private int zkSessionTimeoutMillis = 30000; + + @Parameter(names = { "-c", "-cluster" }, description = "Cluster name") + private String cluster; + + @Parameter(names = { "-cs", "--configuration-store" }, description = "Configuration Store connection string") + private String configurationStore; + + @Parameter(names = { "-h", "--help" }, description = "Show this help message") + private boolean help = false; + } + + public static void main(String[] args) throws InterruptedException { + Arguments arguments = new Arguments(); + JCommander jcommander = new JCommander(); + try { + jcommander.addObject(arguments); + jcommander.parse(args); + if (arguments.help) { + jcommander.usage(); + return; + } + } catch (Exception e) { + jcommander.usage(); + throw e; + } + + ZooKeeper localZk = initZk(arguments.zookeeper, arguments.zkSessionTimeoutMillis); + + deleteZkNodeRecursively(localZk, "/bookies"); + deleteZkNodeRecursively(localZk, "/counters"); + deleteZkNodeRecursively(localZk, "/loadbalance"); + deleteZkNodeRecursively(localZk, "/managed-ledgers"); + deleteZkNodeRecursively(localZk, "/namespace"); + deleteZkNodeRecursively(localZk, "/schemas"); + deleteZkNodeRecursively(localZk, "/stream"); + + if (arguments.configurationStore != null && arguments.cluster != null) { + // Should it be done by REST API before broker is down? + ZooKeeper configStoreZk = initZk(arguments.configurationStore, arguments.zkSessionTimeoutMillis); + deleteZkNodeRecursively(configStoreZk, "/admin/clusters/" + arguments.cluster); + } + + log.info("Cluster metadata for '{}' teardown.", arguments.cluster); + } + + public static ZooKeeper initZk(String connection, int sessionTimeout) throws InterruptedException { + ZooKeeperClientFactory zkFactory = new ZookeeperClientFactoryImpl(); + try { + return zkFactory.create(connection, ZooKeeperClientFactory.SessionType.ReadWrite, sessionTimeout).get(); + } catch (ExecutionException e) { + log.error("Failed to connect to '{}': {}", connection, e.getMessage()); + throw new RuntimeException(e); + } + } + + public static void deleteZkNodeRecursively(ZooKeeper zooKeeper, String path) throws InterruptedException { + try { + ZKUtil.deleteRecursive(zooKeeper, path); + } catch (KeeperException e) { + log.warn("Failed to delete node {} from ZK [{}]: {}", path, zooKeeper, e); + } + } + + private static final Logger log = LoggerFactory.getLogger(PulsarClusterMetadataTeardown.class); +}