@Override
    public boolean format() throws Exception {
        // Clear underreplicated ledgers
        store.deleteRecursive(PulsarLedgerUnderreplicationManager.getBasePath(ledgersRootPath)
                              + BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH)
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);

        // Clear underreplicatedledger locks
        store.deleteRecursive(PulsarLedgerUnderreplicationManager.getUrLockPath(ledgersRootPath))
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);

        // Clear the cookies
        store.deleteRecursive(cookiePath).get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);

        // Clear the INSTANCEID
        if (store.exists(ledgersRootPath + "/" + BookKeeperConstants.INSTANCEID)
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS)) {
            store.delete(ledgersRootPath + "/" + BookKeeperConstants.INSTANCEID, Optional.empty())
                    .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
        }

        // create INSTANCEID
        String instanceId = UUID.randomUUID().toString();
        store.put(ledgersRootPath + "/" + BookKeeperConstants.INSTANCEID,
                instanceId.getBytes(StandardCharsets.UTF_8), Optional.of(-1L))
                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);

        log.info("Successfully formatted BookKeeper metadata");
        return true;
    }
@Test(dataProvider = "impl")
    public void testFormatNonExistingCluster(String provider, Supplier<String> urlSupplier) throws Exception {
        methodSetup(urlSupplier);
        assertClusterNotExists();
        assertTrue(registrationManager.format());
        assertClusterExists();
    }