@Override
    public boolean nukeExistingCluster() throws Exception {
        log.info("Nuking metadata of existing cluster, ledger root path: {}", ledgersRootPath);

        if (!store.exists(ledgersRootPath + "/" + INSTANCEID).get(BLOCKING_CALL_TIMEOUT, MILLISECONDS)) {
            log.info("There is no existing cluster with ledgersRootPath: {}, so exiting nuke operation",
                    ledgersRootPath);
            return true;
        }

        @Cleanup
        RegistrationClient registrationClient = new PulsarRegistrationClient(store, ledgersRootPath);

        Collection<BookieId> rwBookies = registrationClient.getWritableBookies()
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS).getValue();
        if (rwBookies != null && !rwBookies.isEmpty()) {
            log.error("Bookies are still up and connected to this cluster, "
                      + "stop all bookies before nuking the cluster");
            return false;
        }

        Collection<BookieId> roBookies = registrationClient.getReadOnlyBookies()
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS).getValue();
        if (roBookies != null && !roBookies.isEmpty()) {
            log.error("Readonly Bookies are still up and connected to this cluster, "
                      + "stop all bookies before nuking the cluster");
            return false;
        }

        LayoutManager layoutManager = new PulsarLayoutManager(store, ledgersRootPath);
        LedgerManagerFactory ledgerManagerFactory = new PulsarLedgerManagerFactory();
        ledgerManagerFactory.initialize(conf, layoutManager, LegacyHierarchicalLedgerManagerFactory.CUR_VERSION);
        return ledgerManagerFactory.validateAndNukeExistingCluster(conf, layoutManager);
    }
@Test(dataProvider = "impl")
    public void testNukeExistingCluster(String provider, Supplier<String> urlSupplier) throws Exception {
        methodSetup(urlSupplier);
        assertTrue(registrationManager.initNewCluster());
        assertClusterExists();
        assertTrue(registrationManager.nukeExistingCluster());
        assertClusterNotExists();
    }