public static void main(String[] args) throws Exception {
        Arguments arguments = new Arguments();
        CommandLine commander = new CommandLine(arguments);
        try {
            commander.parseArgs(args);
            if (arguments.help) {
                commander.usage(commander.getOut());
                return;
            }
            if (arguments.generateDocs) {
                CmdGenerateDocs cmd = new CmdGenerateDocs("pulsar");
                cmd.addCommand("delete-cluster-metadata", commander);
                cmd.run(null);
                return;
            }
        } catch (Exception e) {
            commander.getErr().println(e);
            throw e;
        }

        @Cleanup
        MetadataStoreExtended metadataStore = MetadataStoreExtended.create(arguments.zookeeper,
                MetadataStoreConfig.builder()
                        .sessionTimeoutMillis(arguments.zkSessionTimeoutMillis)
                        .metadataStoreName(MetadataStoreConfig.METADATA_STORE)
                        .build());

        if (arguments.bkMetadataServiceUri != null) {
            @Cleanup
            BookKeeper bookKeeper =
                    new BookKeeper(new ClientConfiguration().setMetadataServiceUri(arguments.bkMetadataServiceUri));

            @Cleanup("shutdown")
            ManagedLedgerFactory managedLedgerFactory = new ManagedLedgerFactoryImpl(metadataStore, bookKeeper);

            deleteManagedLedgers(metadataStore, managedLedgerFactory);
            deleteSchemaLedgers(metadataStore, bookKeeper);
        }

        for (String localZkNode : localZkNodes) {
            deleteRecursively(metadataStore, "/" + localZkNode).join();
        }

        if (arguments.configurationStore != null && arguments.cluster != null) {
            // Should it be done by REST API before broker is down?
            @Cleanup
            MetadataStore configMetadataStore = MetadataStoreFactory.create(arguments.configurationStore,
                    MetadataStoreConfig.builder().sessionTimeoutMillis(arguments.zkSessionTimeoutMillis)
                            .metadataStoreName(MetadataStoreConfig.CONFIGURATION_METADATA_STORE).build());
            deleteRecursively(configMetadataStore, "/admin/clusters/" + arguments.cluster).join();
        }

        log.info("Cluster metadata for '{}' teardown.", arguments.cluster);
    }
@Test
    public void testMainGenerateDocs() throws Exception {
        PrintStream oldStream = System.out;
        try {
            ByteArrayOutputStream baoStream = new ByteArrayOutputStream();
            System.setOut(new PrintStream(baoStream));

            Class argumentsClass =
                    Class.forName("org.apache.pulsar.PulsarClusterMetadataTeardown$Arguments");

            PulsarClusterMetadataTeardown.main(new String[]{"-zk", "zk", "-g"});

            String message = baoStream.toString();

            Field[] fields = argumentsClass.getDeclaredFields();
            for (Field field : fields) {
                boolean fieldHasAnno = field.isAnnotationPresent(Option.class);
                if (fieldHasAnno) {
                    Option fieldAnno = field.getAnnotation(Option.class);
                    String[] names = fieldAnno.names();
                    if (names.length == 0) {
                        continue;
                    }
                    String nameStr = Arrays.asList(names).toString();
                    nameStr = nameStr.substring(1, nameStr.length() - 1);
                    assertTrue(message.indexOf(nameStr) > 0);
                }
            }
        } finally {
            System.setOut(oldStream);
        }
    }