@Override
    public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(BookieId bookieId) {
        // this method cannot perform blocking calls to the MetadataStore
        // or return a CompletableFuture that is completed on the MetadataStore main thread
        // this is because there are a few cases in which some operations on the main thread
        // wait for the result. This is due to the fact that resolving the address of a bookie
        // is needed in many code paths.
        Versioned<BookieServiceInfo> info;
        if ((info = writableBookieInfo.get(bookieId)) == null) {
            info = readOnlyBookieInfo.get(bookieId);
        }
        if (log.isDebugEnabled()) {
            log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
        }
        if (info != null) {
            return completedFuture(info);
        } else {
            return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
        }
    }
@Test
    public void testNetworkDelayWithBkZkManager() throws Throwable {
        final String zksConnectionString = zks.getConnectionString();
        final String ledgersRoot = "/test/ledgers-" + UUID.randomUUID();
        // prepare registration manager
        @Cleanup
        ZooKeeper zk = new ZooKeeper(zksConnectionString, 5000, null);
        final ServerConfiguration serverConfiguration = new ServerConfiguration();
        serverConfiguration.setZkLedgersRootPath(ledgersRoot);
        final FaultInjectableZKRegistrationManager rm = new FaultInjectableZKRegistrationManager(serverConfiguration, zk);
        rm.prepareFormat();
        // prepare registration client
        @Cleanup
        MetadataStoreExtended store = MetadataStoreExtended.create(zksConnectionString,
                MetadataStoreConfig.builder().fsyncEnable(false).build());
        @Cleanup
        RegistrationClient rc1 = new PulsarRegistrationClient(store, ledgersRoot);
        @Cleanup
        RegistrationClient rc2 = new PulsarRegistrationClient(store, ledgersRoot);

        final List<BookieId> addresses = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            addresses.add(BookieId.parse("BOOKIE-" + i));
        }
        final Map<BookieId, BookieServiceInfo> bookieServiceInfos = new HashMap<>();

        int port = 223;
        for (BookieId address : addresses) {
            BookieServiceInfo info = new BookieServiceInfo();
            BookieServiceInfo.Endpoint endpoint = new BookieServiceInfo.Endpoint();
            endpoint.setAuth(Collections.emptyList());
            endpoint.setExtensions(Collections.emptyList());
            endpoint.setId("id");
            endpoint.setHost("localhost");
            endpoint.setPort(port++);
            endpoint.setProtocol("bookie-rpc");
            info.setEndpoints(Arrays.asList(endpoint));
            bookieServiceInfos.put(address, info);
            rm.registerBookie(address, false, info);
            // write the cookie
            rm.writeCookie(address, new Versioned<>(new byte[0], Version.NEW));
        }

        // trigger loading the BookieServiceInfo in the local cache
        getAndVerifyAllBookies(rc1, addresses);
        getAndVerifyAllBookies(rc2, addresses);

        Awaitility.await().untilAsserted(() -> {
            for (BookieId address : addresses) {
                compareBookieServiceInfo(rc1.getBookieServiceInfo(address).get().getValue(),
                        bookieServiceInfos.get(address));
                compareBookieServiceInfo(rc2.getBookieServiceInfo(address).get().getValue(),
                        bookieServiceInfos.get(address));
            }
        });

        // verified the init status.


        // mock network delay
        rm.betweenRegisterReadOnlyBookie(__ -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return null;
        });

        for (int i = 0; i < addresses.size() / 2; i++) {
            final BookieId bkId = addresses.get(i);
            // turn some bookies to be read only.
            rm.registerBookie(bkId, true, bookieServiceInfos.get(bkId));
        }

        Awaitility.await().untilAsserted(() -> {
            for (BookieId address : addresses) {
                compareBookieServiceInfo(rc1.getBookieServiceInfo(address).get().getValue(),
                        bookieServiceInfos.get(address));
                compareBookieServiceInfo(rc2.getBookieServiceInfo(address).get().getValue(),
                        bookieServiceInfos.get(address));
            }
        });
    }