@Override
    public void generateLedgerId(BookkeeperInternalCallbacks.GenericCallback<Long> genericCallback) {
        ledgerIdGenPathPresent()
                .thenCompose(isIdGenPathPresent -> {
                    if (isIdGenPathPresent) {
                        // We've already started generating 63-bit ledger IDs.
                        // Keep doing that.
                        return generateLongLedgerId();
                    } else {
                        // We've not moved onto 63-bit ledgers yet.
                        return generateShortLedgerId();
                    }
                }).thenAccept(ledgerId ->
                genericCallback.operationComplete(BKException.Code.OK, ledgerId)
        ).exceptionally(ex -> {
            log.error("Error generating ledger id: {}", ex.getMessage());
            genericCallback.operationComplete(BKException.Code.MetaStoreException, -1L);
            return null;
        });

    }
@Test(dataProvider = "impl")
    public void testEnsureCounterIsNotResetWithContainerNodes(String provider, Supplier<String> urlSupplier)
            throws Exception {
        @Cleanup
        MetadataStoreExtended store =
                MetadataStoreExtended.create(urlSupplier.get(), MetadataStoreConfig.builder().build());

        @Cleanup
        PulsarLedgerIdGenerator ledgerIdGenerator = new PulsarLedgerIdGenerator(store, "/ledgers");

        CountDownLatch l1 = new CountDownLatch(1);
        AtomicLong res1 = new AtomicLong();
        ledgerIdGenerator.generateLedgerId((rc, result) -> {
            assertEquals(rc, BKException.Code.OK);
            res1.set(result);
            l1.countDown();
        });

        l1.await();
        log.info("res1 : {}", res1);

        zks.checkContainers();

        CountDownLatch l2 = new CountDownLatch(1);
        AtomicLong res2 = new AtomicLong();
        ledgerIdGenerator.generateLedgerId((rc, result) -> {
            assertEquals(rc, BKException.Code.OK);
            res2.set(result);
            l2.countDown();
        });
        l2.await();

        log.info("res2 : {}", res2);
        assertNotEquals(res1, res2);
        assertTrue(res1.get() < res2.get());
    }