BookkeeperSchemaStorage.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.broker.service.schema;

import static com.google.common.collect.Lists.newArrayList;
import static com.google.protobuf.ByteString.copyFrom;
import static java.util.Objects.isNull;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.Functions.newSchemaEntry;
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import javax.validation.constraints.NotNull;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.LocatorEntry;
import org.apache.pulsar.broker.service.schema.exceptions.SchemaException;
import org.apache.pulsar.common.protocol.schema.SchemaStorage;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.protocol.schema.StoredSchema;
import org.apache.pulsar.common.schema.LongSchemaVersion;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataCache;
import org.apache.pulsar.metadata.api.MetadataSerde;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.Stat;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BookkeeperSchemaStorage implements SchemaStorage<LocatorEntry> {
    private static final Logger log = LoggerFactory.getLogger(BookkeeperSchemaStorage.class);

    private static final String SchemaPath = "/schemas";
    private static final byte[] LedgerPassword = "".getBytes();

    private final MetadataStoreExtended store;
    private final PulsarService pulsar;
    private final MetadataCache<SchemaStorageFormat.SchemaLocator> locatorEntryCache;

    private final ServiceConfiguration config;
    private BookKeeper bookKeeper;

    private final ConcurrentMap<String, CompletableFuture<StoredSchema>> readSchemaOperations =
            new ConcurrentHashMap<>();

    @VisibleForTesting
    BookkeeperSchemaStorage(PulsarService pulsar) {
        this.pulsar = pulsar;
        this.store = pulsar.getLocalMetadataStore();
        this.config = pulsar.getConfiguration();
        this.locatorEntryCache = store.getMetadataCache(new MetadataSerde<SchemaStorageFormat.SchemaLocator>() {
            @Override
            public byte[] serialize(String path, SchemaStorageFormat.SchemaLocator value) {
                return value.toByteArray();
            }

            @Override
            public SchemaStorageFormat.SchemaLocator deserialize(String path, byte[] content, Stat stat)
                    throws IOException {
                return SchemaStorageFormat.SchemaLocator.parseFrom(content);
            }
        });
    }

    @Override
    public void start() throws IOException {
        this.bookKeeper = pulsar.getBookKeeperClientFactory().create(
            pulsar.getConfiguration(),
            store,
            pulsar.getIoEventLoopGroup(),
            Optional.empty(),
            null
        );
    }

    @Override
    public CompletableFuture<SchemaVersion> put(String key, byte[] value, byte[] hash) {
        return putSchema(key, value, hash).thenApply(LongSchemaVersion::new);
    }

    public CompletableFuture<SchemaVersion> put(String key, byte[] value, byte[] hash, LocatorEntry locatorEntry) {
        return putSchema(key, value, hash, locatorEntry).thenApply(LongSchemaVersion::new);
    }

    @Override
    public CompletableFuture<StoredSchema> get(String key, SchemaVersion version) {
        if (version == SchemaVersion.Latest) {
            return getSchema(key);
        } else {
            LongSchemaVersion longVersion = (LongSchemaVersion) version;
            return getSchema(key, longVersion.getVersion());
        }
    }

    @Override
    public CompletableFuture<List<CompletableFuture<StoredSchema>>> getAll(String key) {
        return getAll(key, null);
    }

    public CompletableFuture<List<CompletableFuture<StoredSchema>>> getAll(String key, LocatorEntry locator) {
        CompletableFuture<List<CompletableFuture<StoredSchema>>> result = new CompletableFuture<>();
        if (log.isDebugEnabled()) {
            log.debug("[{}] Get all schemas - locator: {}", key, locator);
        }
        CompletableFuture<Optional<LocatorEntry>> locatorFuture = locator == null
                ? getLocator(key) : CompletableFuture.completedFuture(Optional.of(locator));
        List<CompletableFuture<StoredSchema>> list = new ArrayList<>();
        locatorFuture.thenAccept(locatorEntry -> {
            if (locatorEntry.isPresent()) {
                locatorEntry.get().locator.getIndexList().forEach(indexEntry ->
                        list.add(readSchemaEntry(indexEntry.getPosition()).thenApply(entry -> new StoredSchema(
                                        entry.getSchemaData().toByteArray(),
                                        new LongSchemaVersion(indexEntry.getVersion())))));
                result.complete(list);
            } else {
                result.complete(Collections.emptyList());
            }
        });
        return result;
    }

    @Override
    public CompletableFuture<Optional<LocatorEntry>> getLocator(String key) {
        return getSchemaLocator(getSchemaPath(key));
    }

    public List<Long> getSchemaLedgerList(String key) throws IOException {
        Optional<LocatorEntry> locatorEntry = null;
        try {
            locatorEntry = getLocator(key).get();
        } catch (Exception e) {
            log.warn("Failed to get list of schema-storage ledger for {}, the exception as follow: \n {}", key,
                    (e instanceof ExecutionException ? e.getCause() : e));
            throw new IOException("Failed to get schema ledger for" + key);
        }
        LocatorEntry entry = locatorEntry.orElse(null);
        return entry != null ? entry.locator.getIndexList().stream().map(i -> i.getPosition().getLedgerId())
                .collect(Collectors.toList()) : null;
    }

    @VisibleForTesting
    BookKeeper getBookKeeper() {
        return bookKeeper;
    }

    @Override
    public CompletableFuture<SchemaVersion> delete(String key, boolean forcefully) {
        return deleteSchema(key, forcefully).thenApply(version -> {
            if (version == null) {
                return null;
            }
            return new LongSchemaVersion(version);
        });
    }

    @Override
    public CompletableFuture<SchemaVersion> delete(String key) {
        return delete(key, false);
    }

    @NotNull
    private CompletableFuture<StoredSchema> getSchema(String schemaId) {
        // There's already a schema read operation in progress. Just piggyback on that
        return readSchemaOperations.computeIfAbsent(schemaId, key -> {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Fetching schema from store", schemaId);
            }
            return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] Got schema locator {}", schemaId, locator);
                }
                if (!locator.isPresent()) {
                    return completedFuture(null);
                }

                SchemaStorageFormat.SchemaLocator schemaLocator = locator.get().locator;

                return readSchemaEntry(schemaLocator.getInfo().getPosition())
                        .thenApply(entry -> new StoredSchema(entry.getSchemaData().toByteArray(),
                                new LongSchemaVersion(schemaLocator.getInfo().getVersion())));
            });
        }).whenComplete((res, ex) -> {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Get operation completed. res={} -- ex={}", schemaId, res, ex);
            }
            readSchemaOperations.remove(schemaId);
        });
    }

    @Override
    public SchemaVersion versionFromBytes(byte[] version) {
        // The schema storage converts the schema from bytes to long
        // so it handles both cases 1) version is 64 bytes long pre 2.4.0;
        // 2) version is 8 bytes long post 2.4.0
        //
        // NOTE: if you are planning to change the logic here. you should consider
        //       both 64 bytes and 8 bytes cases.
        ByteBuffer bb = ByteBuffer.wrap(version);
        return new LongSchemaVersion(bb.getLong());
    }

    @Override
    public void close() throws Exception {
        if (bookKeeper != null) {
            bookKeeper.close();
        }
    }

    @NotNull
    private CompletableFuture<StoredSchema> getSchema(String schemaId, long version) {
        if (log.isDebugEnabled()) {
            log.debug("[{}] Get schema - version: {}", schemaId, version);
        }

        return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Get schema - version: {} - locator: {}", schemaId, version, locator);
            }

            if (!locator.isPresent()) {
                return completedFuture(null);
            }

            SchemaStorageFormat.SchemaLocator schemaLocator = locator.get().locator;
            if (version > schemaLocator.getInfo().getVersion()) {
                return completedFuture(null);
            }

            return findSchemaEntryByVersion(schemaLocator.getIndexList(), version)
                .thenApply(entry ->
                        new StoredSchema(
                            entry.getSchemaData().toByteArray(),
                            new LongSchemaVersion(version)
                        )
                );
        });
    }

    @NotNull
    private CompletableFuture<Long> putSchema(String schemaId, byte[] data, byte[] hash) {
        return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry -> {

            if (optLocatorEntry.isPresent()) {

                SchemaStorageFormat.SchemaLocator locator = optLocatorEntry.get().locator;

                if (log.isDebugEnabled()) {
                    log.debug("[{}] findSchemaEntryByHash - hash={}", schemaId, hash);
                }

                //don't check the schema whether already exist
                return readSchemaEntry(locator.getIndexList().get(0).getPosition())
                        .thenCompose(schemaEntry -> addNewSchemaEntryToStore(schemaId,
                                locator.getIndexList(), data).thenCompose(
                                position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash))
                );
            } else {
                // No schema was defined yet
                return createNewSchema(schemaId, data, hash);
            }
        });
    }

    @NotNull
    private CompletableFuture<Long> putSchema(String schemaId, byte[] data, byte[] hash, LocatorEntry locatorEntry) {
        if (locatorEntry == null) {
            return createNewSchema(schemaId, data, hash);
        } else {
            return addNewSchemaEntryToStore(schemaId,
                    locatorEntry.locator.getIndexList(), data).thenCompose(
                    position -> updateSchemaLocator(schemaId, locatorEntry, position, hash));
        }
    }

    private CompletableFuture<Long> createNewSchema(String schemaId, byte[] data, byte[] hash) {
        SchemaStorageFormat.IndexEntry emptyIndex = SchemaStorageFormat.IndexEntry.newBuilder()
                        .setVersion(0)
                        .setHash(copyFrom(hash))
                        .setPosition(SchemaStorageFormat.PositionInfo.newBuilder()
                                .setEntryId(-1L)
                                .setLedgerId(-1L)
                        ).build();

        return addNewSchemaEntryToStore(schemaId, Collections.singletonList(emptyIndex), data).thenCompose(position -> {
            // The schema was stored in the ledger, now update the z-node with the pointer to it
            SchemaStorageFormat.IndexEntry info = SchemaStorageFormat.IndexEntry.newBuilder()
                    .setVersion(0)
                    .setPosition(position)
                    .setHash(copyFrom(hash))
                    .build();

            return createSchemaLocator(getSchemaPath(schemaId), SchemaStorageFormat.SchemaLocator.newBuilder()
                    .setInfo(info)
                    .addAllIndex(
                            newArrayList(info))
                    .build())
                            .thenApply(ignore -> 0L);
        });
    }

    @NotNull
    private CompletableFuture<Long> deleteSchema(String schemaId, boolean forcefully) {
        return (forcefully ? CompletableFuture.completedFuture(null)
                : ignoreUnrecoverableBKException(getSchema(schemaId))).thenCompose(schemaAndVersion -> {
            if (!forcefully && isNull(schemaAndVersion)) {
                return completedFuture(null);
            } else {
                // The version is only for the compatibility of the current interface
                final long version = -1;
                CompletableFuture<Long> future = new CompletableFuture<>();
                getLocator(schemaId).whenComplete((locator, ex) -> {
                    if (ex != null) {
                        future.completeExceptionally(ex);
                    } else {
                        if (!locator.isPresent()) {
                            future.complete(null);
                            return;
                        }
                        List<SchemaStorageFormat.IndexEntry> indexEntryList = locator.get().locator.getIndexList();
                        List<CompletableFuture<Void>> deleteFutures = new ArrayList<>(indexEntryList.size());
                        indexEntryList.forEach(indexEntry -> {
                            final long ledgerId = indexEntry.getPosition().getLedgerId();
                            CompletableFuture<Void> deleteFuture = new CompletableFuture<>();
                            deleteFutures.add(deleteFuture);
                            bookKeeper.asyncDeleteLedger(ledgerId, (int rc, Object cnx) -> {
                                if (rc != BKException.Code.OK) {
                                    // It's not a serious error, we didn't need call future.completeExceptionally()
                                    log.warn("Failed to delete ledger {} of {}: {}", ledgerId, schemaId, rc);
                                }
                                deleteFuture.complete(null);
                            }, null);
                        });
                        FutureUtil.waitForAll(deleteFutures).whenComplete((v, e) -> {
                            final String path = getSchemaPath(schemaId);
                            store.delete(path, Optional.empty())
                                    .thenRun(() -> {
                                        future.complete(version);
                                    }).exceptionally(zkException -> {
                                        if (zkException.getCause()
                                                instanceof MetadataStoreException.NotFoundException) {
                                            // The znode has been deleted by others.
                                            // In some cases, the program may enter this logic.
                                            // Since the znode is gone, we don’t need to deal with it.
                                            if (log.isDebugEnabled()) {
                                                log.debug("No node for schema path: {}", path);
                                            }
                                            future.complete(null);
                                        } else {
                                            future.completeExceptionally(zkException);
                                        }
                                        return null;
                            });
                        });
                    }
                });
                return future;
            }
        });
    }

    @NotNull
    private static String getSchemaPath(String schemaId) {
        return SchemaPath + "/" + schemaId;
    }

    @NotNull
    private CompletableFuture<SchemaStorageFormat.PositionInfo> addNewSchemaEntryToStore(
        String schemaId,
        List<SchemaStorageFormat.IndexEntry> index,
        byte[] data
    ) {
        SchemaStorageFormat.SchemaEntry schemaEntry = newSchemaEntry(index, data);
        return createLedger(schemaId).thenCompose(ledgerHandle -> {
            final long ledgerId = ledgerHandle.getId();
            return addEntry(ledgerHandle, schemaEntry)
                    .thenApply(entryId -> {
                        ledgerHandle.closeAsync();
                        return Functions.newPositionInfo(ledgerId, entryId);
                    });
        });
    }

    @NotNull
    private CompletableFuture<Long> updateSchemaLocator(
        String schemaId,
        LocatorEntry locatorEntry,
        SchemaStorageFormat.PositionInfo position,
        byte[] hash
    ) {
        long nextVersion = locatorEntry.locator.getInfo().getVersion() + 1;
        SchemaStorageFormat.SchemaLocator locator = locatorEntry.locator;
        SchemaStorageFormat.IndexEntry info =
            SchemaStorageFormat.IndexEntry.newBuilder()
                .setVersion(nextVersion)
                .setPosition(position)
                .setHash(copyFrom(hash))
                .build();

        final ArrayList<SchemaStorageFormat.IndexEntry> indexList = new ArrayList<>();
        indexList.addAll(locator.getIndexList());
        indexList.add(info);
        return updateSchemaLocator(getSchemaPath(schemaId),
                SchemaStorageFormat.SchemaLocator.newBuilder()
                        .setInfo(info)
                        .addAllIndex(indexList)
                        .build()
                , locatorEntry.version
        ).thenApply(ignore -> nextVersion);
    }

    @NotNull
    private CompletableFuture<SchemaStorageFormat.SchemaEntry> findSchemaEntryByVersion(
        List<SchemaStorageFormat.IndexEntry> index,
        long version
    ) {

        if (index.isEmpty()) {
            return completedFuture(null);
        }

        SchemaStorageFormat.IndexEntry lowest = index.get(0);
        if (version < lowest.getVersion()) {
            return readSchemaEntry(lowest.getPosition())
                    .thenCompose(entry -> findSchemaEntryByVersion(entry.getIndexList(), version));
        }

        for (SchemaStorageFormat.IndexEntry entry : index) {
            if (entry.getVersion() == version) {
                return readSchemaEntry(entry.getPosition());
            } else if (entry.getVersion() > version) {
                break;
            }
        }

        return completedFuture(null);
    }

    @NotNull
    private CompletableFuture<SchemaStorageFormat.SchemaEntry> readSchemaEntry(
        SchemaStorageFormat.PositionInfo position
    ) {
        if (log.isDebugEnabled()) {
            log.debug("Reading schema entry from {}", position);
        }

        return openLedger(position.getLedgerId())
            .thenCompose((ledger) ->
                Functions.getLedgerEntry(ledger, position.getEntryId())
                    .thenCompose(entry -> closeLedger(ledger)
                        .thenApply(ignore -> entry)
                    )
            ).thenCompose(Functions::parseSchemaEntry);
    }

    @NotNull
    private CompletableFuture<Void> updateSchemaLocator(String id,
                                                        SchemaStorageFormat.SchemaLocator schema, long version) {
        return store.put(id, schema.toByteArray(), Optional.of(version)).thenApply(__ -> null);
    }

    @NotNull
    private CompletableFuture<LocatorEntry> createSchemaLocator(String id, SchemaStorageFormat.SchemaLocator locator) {
        return store.put(id, locator.toByteArray(), Optional.of(-1L))
                .thenApply(stat -> new LocatorEntry(locator, stat.getVersion()));
    }

    @NotNull
    private CompletableFuture<Optional<LocatorEntry>> getSchemaLocator(String schema) {
        return locatorEntryCache.getWithStats(schema)
                .thenApply(o ->
                        o.map(r -> new LocatorEntry(r.getValue(), r.getStat().getVersion())));
    }

    @NotNull
    private CompletableFuture<Long> addEntry(LedgerHandle ledgerHandle, SchemaStorageFormat.SchemaEntry entry) {
        final CompletableFuture<Long> future = new CompletableFuture<>();
        ledgerHandle.asyncAddEntry(entry.toByteArray(),
            (rc, handle, entryId, ctx) -> {
                if (rc != BKException.Code.OK) {
                    future.completeExceptionally(bkException("Failed to add entry", rc, ledgerHandle.getId(), -1));
                } else {
                    future.complete(entryId);
                }
            }, null
        );
        return future;
    }

    @NotNull
    private CompletableFuture<LedgerHandle> createLedger(String schemaId) {
        Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForSchema(schemaId);
        final CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
        try {
            bookKeeper.asyncCreateLedger(
                    config.getManagedLedgerDefaultEnsembleSize(),
                    config.getManagedLedgerDefaultWriteQuorum(),
                    config.getManagedLedgerDefaultAckQuorum(),
                    BookKeeper.DigestType.fromApiDigestType(config.getManagedLedgerDigestType()),
                    LedgerPassword,
                    (rc, handle, ctx) -> {
                        if (rc != BKException.Code.OK) {
                            future.completeExceptionally(bkException("Failed to create ledger", rc, -1, -1));
                        } else {
                            future.complete(handle);
                        }
                    }, null, metadata);
        } catch (Throwable t) {
            log.error("[{}] Encountered unexpected error when creating schema ledger", schemaId, t);
            return FutureUtil.failedFuture(t);
        }
        return future;
    }

    @NotNull
    private CompletableFuture<LedgerHandle> openLedger(Long ledgerId) {
        final CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
        bookKeeper.asyncOpenLedger(
            ledgerId,
            BookKeeper.DigestType.fromApiDigestType(config.getManagedLedgerDigestType()),
            LedgerPassword,
            (rc, handle, ctx) -> {
                if (rc != BKException.Code.OK) {
                    future.completeExceptionally(bkException("Failed to open ledger", rc, ledgerId, -1));
                } else {
                    future.complete(handle);
                }
            }, null
        );
        return future;
    }

    @NotNull
    private CompletableFuture<Void> closeLedger(LedgerHandle ledgerHandle) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        ledgerHandle.asyncClose((rc, handle, ctx) -> {
            if (rc != BKException.Code.OK) {
                future.completeExceptionally(bkException("Failed to close ledger", rc, ledgerHandle.getId(), -1));
            } else {
                future.complete(null);
            }
        }, null);
        return future;
    }

    public CompletableFuture<List<Long>> getStoreLedgerIdsBySchemaId(String schemaId) {
        CompletableFuture<List<Long>> ledgerIdsFuture = new CompletableFuture<>();
        getSchemaLocator(getSchemaPath(schemaId)).thenAccept(locator -> {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Get all store schema ledgerIds - locator: {}", schemaId, locator);
            }

            if (!locator.isPresent()) {
                ledgerIdsFuture.complete(Collections.emptyList());
                return;
            }
            Set<Long> ledgerIds = new HashSet<>();
            SchemaStorageFormat.SchemaLocator schemaLocator = locator.get().locator;
            schemaLocator.getIndexList().forEach(indexEntry -> ledgerIds.add(indexEntry.getPosition().getLedgerId()));
            ledgerIdsFuture.complete(new ArrayList<>(ledgerIds));
        }).exceptionally(e -> {
            ledgerIdsFuture.completeExceptionally(e);
            return null;
        });
        return ledgerIdsFuture;
    }

    interface Functions {
        static CompletableFuture<LedgerEntry> getLedgerEntry(LedgerHandle ledger, long entry) {
            final CompletableFuture<LedgerEntry> future = new CompletableFuture<>();
            ledger.asyncReadEntries(entry, entry,
                (rc, handle, entries, ctx) -> {
                    if (rc != BKException.Code.OK) {
                        future.completeExceptionally(bkException("Failed to read entry", rc, ledger.getId(), entry));
                    } else {
                        future.complete(entries.nextElement());
                    }
                }, null
            );
            return future;
        }

        static CompletableFuture<SchemaStorageFormat.SchemaEntry> parseSchemaEntry(LedgerEntry ledgerEntry) {
            CompletableFuture<SchemaStorageFormat.SchemaEntry> result = new CompletableFuture<>();
            try {
                result.complete(SchemaStorageFormat.SchemaEntry.parseFrom(ledgerEntry.getEntry()));
            } catch (IOException e) {
                result.completeExceptionally(e);
            }
            return result;
        }

        static SchemaStorageFormat.SchemaEntry newSchemaEntry(
            List<SchemaStorageFormat.IndexEntry> index,
            byte[] data
        ) {
            return SchemaStorageFormat.SchemaEntry.newBuilder()
                .setSchemaData(copyFrom(data))
                .addAllIndex(index)
                .build();
        }

        static SchemaStorageFormat.PositionInfo newPositionInfo(long ledgerId, long entryId) {
            return SchemaStorageFormat.PositionInfo.newBuilder()
                .setLedgerId(ledgerId)
                .setEntryId(entryId)
                .build();
        }
    }

    public static class LocatorEntry {
        final SchemaStorageFormat.SchemaLocator locator;
        final long version;

        LocatorEntry(SchemaStorageFormat.SchemaLocator locator, long version) {
            this.locator = locator;
            this.version = version;
        }
    }

    public static Exception bkException(String operation, int rc, long ledgerId, long entryId) {
        String message = org.apache.bookkeeper.client.api.BKException.getMessage(rc)
                + " -  ledger=" + ledgerId + " - operation=" + operation;

        if (entryId != -1) {
            message += " - entry=" + entryId;
        }
        boolean recoverable = rc != BKException.Code.NoSuchLedgerExistsException
                && rc != BKException.Code.NoSuchEntryException;
        return new SchemaException(recoverable, message);
    }

    public static <T> CompletableFuture<T> ignoreUnrecoverableBKException(CompletableFuture<T> source) {
        return source.exceptionally(t -> {
            if (t.getCause() != null
                    && (t.getCause() instanceof SchemaException)
                    && !((SchemaException) t.getCause()).isRecoverable()) {
                // Meeting NoSuchLedgerExistsException or NoSuchEntryException when reading schemas in
                // bookkeeper. This also means that the data has already been deleted by other operations
                // in deleting schema.
                if (log.isDebugEnabled()) {
                    log.debug("Schema data in bookkeeper may be deleted by other operations.", t);
                }
                return null;
            }
            // rethrow other cases
            throw t instanceof CompletionException ? (CompletionException) t : new CompletionException(t);
        });
    }
}