@Override
    public void scanLedgers(OffloadedLedgerMetadataConsumer consumer, Map<String,
            String> offloadDriverMetadata) throws ManagedLedgerException {
        BlobStoreLocation bsKey = getBlobStoreLocation(offloadDriverMetadata);
        String endpoint = bsKey.getEndpoint();
        String readBucket = bsKey.getBucket();
        log.info("Scanning bucket {}, bsKey {}, location {} endpoint{} ", readBucket, bsKey,
                config.getBlobStoreLocation(), endpoint);
        BlobStore readBlobstore = getBlobStore(config.getBlobStoreLocation());
        int batchSize = 100;
        String bucketName = config.getBucket();
        String marker = null;
        do {
            marker = scanContainer(consumer, readBlobstore, bucketName, marker, batchSize);
        } while (marker != null);

    }
@Test(timeOut = 600000)  // 10 minutes.
    public void testScanLedgers() throws Exception {
        ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3);
        LedgerOffloader offloader = getOffloader();

        UUID uuid = UUID.randomUUID();
        offloader.offload(toWrite, uuid, new HashMap<>()).get();

        List<OffloadedLedgerMetadata> result = new ArrayList<>();
        offloader.scanLedgers(
                (m) -> {
                    log.info("found {}", m);
                    if (m.getLedgerId() == toWrite.getId()) {
                        result.add(m);
                    }
                    return true;
                }, offloader.getOffloadDriverMetadata());
        assertEquals(2, result.size());

        // data and index

        OffloadedLedgerMetadata offloadedLedgerMetadata = result.get(0);
        assertEquals(toWrite.getId(), offloadedLedgerMetadata.getLedgerId());

        OffloadedLedgerMetadata offloadedLedgerMetadata2 = result.get(1);
        assertEquals(toWrite.getId(), offloadedLedgerMetadata2.getLedgerId());
    }