private LedgerMetadata createLedgerMetadata() throws Exception {

        Map<String, byte[]> metadataCustom = Maps.newHashMap();
        metadataCustom.put("key1", "value1".getBytes(UTF_8));
        metadataCustom.put("key7", "value7".getBytes(UTF_8));

        ArrayList<BookieSocketAddress> bookies = Lists.newArrayList();
        BookieSocketAddress BOOKIE1 = new BookieSocketAddress("127.0.0.1:3181");
        BookieSocketAddress BOOKIE2 = new BookieSocketAddress("127.0.0.2:3181");
        BookieSocketAddress BOOKIE3 = new BookieSocketAddress("127.0.0.3:3181");
        bookies.add(0, BOOKIE1);
        bookies.add(1, BOOKIE2);
        bookies.add(2, BOOKIE3);

        return LedgerMetadataBuilder.create().withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
            .withDigestType(DigestType.CRC32C).withPassword("password".getBytes(UTF_8))
            .withCustomMetadata(metadataCustom).withClosedState().withLastEntryId(5000).withLength(100)
            .newEnsembleEntry(0L, bookies).build();
    }
 

@Override
public void asyncClose(CloseCallback cb, Object ctx) {
    bk.getProgrammedFailure().thenComposeAsync((res) -> {
            fenced = true;

            Versioned<LedgerMetadata> current = getVersionedLedgerMetadata();
            Versioned<LedgerMetadata> newMetadata = new Versioned<>(
                    LedgerMetadataBuilder.from(current.getValue())
                    .withClosedState().withLastEntryId(getLastAddConfirmed())
                    .withLength(getLength()).build(),
                    new LongVersion(((LongVersion)current.getVersion()).getLongVersion() + 1));
            setLedgerMetadata(current, newMetadata);
            return FutureUtils.value(null);
        }, bk.executor).whenCompleteAsync((res, exception) -> {
                if (exception != null) {
                    cb.closeComplete(PulsarMockBookKeeper.getExceptionCode(exception), null, ctx);
                } else {
                    cb.closeComplete(BKException.Code.OK, this, ctx);
                }
            }, bk.executor);
}
 

public static OffloadIndexBlockImpl get(LedgerMetadata metadata, long dataObjectLength,
                                        long dataHeaderLength,
                                        List<OffloadIndexEntryImpl> entries) {
    OffloadIndexBlockImpl block = RECYCLER.get();
    block.indexEntries = Maps.newTreeMap();
    entries.forEach(entry -> block.indexEntries.putIfAbsent(entry.getEntryId(), entry));
    checkState(entries.size() == block.indexEntries.size());
    block.segmentMetadata = metadata;
    block.dataObjectLength = dataObjectLength;
    block.dataHeaderLength = dataHeaderLength;
    return block;
}
 

private void assertAffinityBookies(LedgerManager ledgerManager, List<LedgerInfo> ledgers1,
        Set<BookieSocketAddress> defaultBookies) throws Exception {
    for (LedgerInfo lInfo : ledgers1) {
        long ledgerId = lInfo.getLedgerId();
        CompletableFuture<Versioned<LedgerMetadata>> ledgerMetaFuture = ledgerManager.readLedgerMetadata(ledgerId);
        LedgerMetadata ledgerMetadata = ledgerMetaFuture.get().getValue();
        Set<BookieSocketAddress> ledgerBookies = Sets.newHashSet();
        ledgerBookies.addAll(ledgerMetadata.getAllEnsembles().values().iterator().next());
        assertEquals(ledgerBookies.size(), defaultBookies.size());
        ledgerBookies.removeAll(defaultBookies);
        assertEquals(ledgerBookies.size(), 0);
    }
}
 

private static LedgerMetadata createMetadata(DigestType digest, byte[] passwd) {
    List<BookieSocketAddress> ensemble = Lists.newArrayList(
            new BookieSocketAddress("192.0.2.1", 1234),
            new BookieSocketAddress("192.0.2.2", 1234),
            new BookieSocketAddress("192.0.2.3", 1234));
    return LedgerMetadataBuilder.create()
        .withDigestType(digest.toApiDigestType())
        .withPassword(passwd)
        .newEnsembleEntry(0L, ensemble)
        .build();
}
 

MockMetadata(LedgerMetadata toCopy) {
    ensembleSize = toCopy.getEnsembleSize();
    writeQuorumSize = toCopy.getWriteQuorumSize();
    ackQuorumSize = toCopy.getAckQuorumSize();
    lastEntryId = toCopy.getLastEntryId();
    length = toCopy.getLength();
    digestType = toCopy.getDigestType();
    ctime = toCopy.getCtime();
    isClosed = toCopy.isClosed();
    metadataFormatVersion = toCopy.getMetadataFormatVersion();
    state = toCopy.getState();
    password = Arrays.copyOf(toCopy.getPassword(), toCopy.getPassword().length);
    customMetadata = ImmutableMap.copyOf(toCopy.getCustomMetadata());
}
 

public static void collectLedgers(BookKeeper bk, long ledgerId, AsyncCallback.VoidCallback cb, Map<Long, LedgerMetadata> ledgers) {
    try {
        bk.getLedgerManager()
          .readLedgerMetadata(ledgerId)
          .thenAccept((v) -> {
              LOG.debug("Got ledger metadata");
              ledgers.put(ledgerId, v.getValue());
          })
          .thenAccept((v) -> {
              cb.processResult(BKException.Code.OK, null, null);
          });
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return index.getLedgerMetadata();
}
 

@Override
public OffloadIndexBlockBuilder withLedgerMetadata(LedgerMetadata metadata) {
    this.ledgerMetadata = metadata;
    return this;
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return this.segmentMetadata;
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return null;
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return ledgerMetadata;

}
 

PulsarMockReadHandle(PulsarMockBookKeeper bk, long ledgerId, LedgerMetadata metadata, List<LedgerEntryImpl> entries) {
    this.bk = bk;
    this.ledgerId = ledgerId;
    this.metadata = metadata;
    this.entries = entries;
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return metadata;
}
 

public static LedgerMetadata parseLedgerMetadata(byte[] bytes) throws IOException {
    DataFormats.LedgerMetadataFormat ledgerMetadataFormat = DataFormats.LedgerMetadataFormat.newBuilder().mergeFrom(bytes).build();
    LedgerMetadataBuilder builder = LedgerMetadataBuilder.create()
            .withLastEntryId(ledgerMetadataFormat.getLastEntryId())
            .withPassword(ledgerMetadataFormat.getPassword().toByteArray())
            .withClosedState()
            .withMetadataFormatVersion(2)
            .withLength(ledgerMetadataFormat.getLength())
            .withAckQuorumSize(ledgerMetadataFormat.getAckQuorumSize())
            .withCreationTime(ledgerMetadataFormat.getCtime())
            .withWriteQuorumSize(ledgerMetadataFormat.getQuorumSize())
            .withEnsembleSize(ledgerMetadataFormat.getEnsembleSize());
    ledgerMetadataFormat.getSegmentList().forEach(segment -> {
        ArrayList<BookieSocketAddress> addressArrayList = new ArrayList<>();
        segment.getEnsembleMemberList().forEach(address -> {
            try {
                addressArrayList.add(new BookieSocketAddress(address));
            } catch (IOException e) {
                log.error("Exception when create BookieSocketAddress. ", e);
            }
        });
        builder.newEnsembleEntry(segment.getFirstEntryId(), addressArrayList);
    });

    if (ledgerMetadataFormat.getCustomMetadataCount() > 0) {
        Map<String, byte[]> customMetadata = Maps.newHashMap();
        ledgerMetadataFormat.getCustomMetadataList().forEach(
                entry -> customMetadata.put(entry.getKey(), entry.getValue().toByteArray()));
        builder.withCustomMetadata(customMetadata);
    }

    switch (ledgerMetadataFormat.getDigestType()) {
        case HMAC:
            builder.withDigestType(DigestType.MAC);
            break;
        case CRC32:
            builder.withDigestType(DigestType.CRC32);
            break;
        case CRC32C:
            builder.withDigestType(DigestType.CRC32C);
            break;
        case DUMMY:
            builder.withDigestType(DigestType.DUMMY);
            break;
        default:
            throw new IllegalArgumentException("Unable to convert digest type " + ledgerMetadataFormat.getDigestType());
    }

    return builder.build();
}
 

@Override
public LedgerMetadata getLedgerMetadata() {
    return metadata;
}
 

/**
 * Iterates over all available ledgers and returns the first one that has
 * a metadata key 'name' equals to the given name
 * @param bk
 * @param name
 * @return
 * @throws Exception
 */
public static Optional<Long> findLedgerByName(BookKeeper bk, String name) throws Exception {
    Map<Long, LedgerMetadata> ledgers = new HashMap<Long, LedgerMetadata>();
    final AtomicInteger returnCode = new AtomicInteger(BKException.Code.OK);
    final CountDownLatch processDone = new CountDownLatch(1);

    // There's no standard "list" operation. Instead, BK offers a generalized way to
    // iterate over all available ledgers using an async visitor callback.
    // The second callback will be called when there are no more ledgers do process or if an
    // error occurs.
    bk.getLedgerManager()
      .asyncProcessLedgers(
        (ledgerId, cb) -> collectLedgers(bk, ledgerId, cb, ledgers),
        (rc, s, obj) -> {
            returnCode.set(rc);
            processDone.countDown();
        }, 
        null, 
        BKException.Code.OK, BKException.Code.ReadException);
    processDone.await(5, TimeUnit.MINUTES);
    LOG.info("Ledgers collected: total found=" + ledgers.size());

    byte[] nameBytes = name.getBytes();
    Optional<Entry<Long, LedgerMetadata>> entry = ledgers.entrySet()
      .stream()
      .filter((e) -> {
          Map<String, byte[]> meta = e.getValue()
            .getCustomMetadata();
          if (meta != null) {
              LOG.info("ledger: " + e.getKey() + ", customMeta=" + meta);
              byte[] data = meta.get("name");
              if (data != null && Arrays.equals(data, nameBytes)) {
                  return true;
              } else {
                  return false;
              }
          } else {
              LOG.info("ledger: " + e.getKey() + ", no meta");
              return false;
          }
      })
      .findFirst();
    if (entry.isPresent()) {
        return Optional.of(entry.get()
            .getKey());
    } else {
        return Optional.empty();
    }
}
 

/**
 * Build index block with the passed in ledger metadata.
 *
 * @param metadata the ledger metadata
 */
OffloadIndexBlockBuilder withLedgerMetadata(LedgerMetadata metadata);
 

/**
 * Get LedgerMetadata.
 */
LedgerMetadata getLedgerMetadata();