@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(dataProvider = "impl")
    public void testGetBookieServiceInfo(String provider, Supplier<String> urlSupplier) throws Exception {
        @Cleanup
        MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
                MetadataStoreConfig.builder().fsyncEnable(false).build());

        String ledgersRoot = "/test/ledgers-" + UUID.randomUUID();

        @Cleanup
        RegistrationManager rm = new PulsarRegistrationManager(store, ledgersRoot, mock(AbstractConfiguration.class));

        @Cleanup
        RegistrationClient rc = new PulsarRegistrationClient(store, ledgersRoot);

        List<BookieId> addresses = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            addresses.add(BookieId.parse("BOOKIE-" + i));
        }
        Map<BookieId, BookieServiceInfo> bookieServiceInfos = new HashMap<>();
        Set<BookieId> readOnlyBookies = new HashSet<>();
        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);
            // some readonly, some writable
            boolean readOnly = port % 2 == 0;
            if (readOnly) {
                readOnlyBookies.add(address);
            }
            rm.registerBookie(address, readOnly, info);
            // write the cookie
            rm.writeCookie(address, new Versioned<>(new byte[0], Version.NEW));
        }

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

        Awaitility.await().untilAsserted(() -> {
            for (BookieId address : addresses) {
                BookieServiceInfo bookieServiceInfo = rc.getBookieServiceInfo(address).get().getValue();
                compareBookieServiceInfo(bookieServiceInfo, bookieServiceInfos.get(address));
            }});

        // shutdown the bookies (but keep the cookie)
        for (BookieId address : addresses) {
            rm.unregisterBookie(address, readOnlyBookies.contains(address));
            readOnlyBookies.remove(address);
        }

        // getAllBookies should find all the bookies in any case (it reads the cookies)
        getAndVerifyAllBookies(rc, addresses);

        // getBookieServiceInfo should fail with BKBookieHandleNotAvailableException
        Awaitility.await().untilAsserted(() -> {
            for (BookieId address : addresses) {
                assertTrue(
                        expectThrows(ExecutionException.class, () -> {
                            rc.getBookieServiceInfo(address).get();
                        }).getCause() instanceof BKException.BKBookieHandleNotAvailableException);
            }});


        // restart the bookies, all writable
        // we 'register' the bookie, but do not write the cookie again
        for (BookieId address : addresses) {
            rm.registerBookie(address, false, bookieServiceInfos.get(address));
        }

        getAndVerifyAllBookies(rc, addresses);

        // verify that infos are available again
        Awaitility.await()
                .ignoreExceptionsMatching(e -> e.getCause() instanceof BKException.BKBookieHandleNotAvailableException)
                .untilAsserted(() -> {
                    for (BookieId address : addresses) {
                        BookieServiceInfo bookieServiceInfo = rc.getBookieServiceInfo(address).get().getValue();
                        compareBookieServiceInfo(bookieServiceInfo, bookieServiceInfos.get(address));
                    }
                });

        // update the infos
        port = 111;
        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);
            // some readonly, some writable
            boolean readOnly = port % 2 == 0;

            // remove the previous info from the metadata service
            rm.unregisterBookie(address, readOnlyBookies.contains(address));

            rm.registerBookie(address, readOnly, info);

            if (readOnly) {
                readOnlyBookies.add(address);
            }
        }

        // verify that the client tracked the changes
        Awaitility
                .await()
                .ignoreExceptionsMatching(e -> e.getCause() instanceof BKException.BKBookieHandleNotAvailableException)
                .untilAsserted(() -> {
                    // verify that infos are updated
                    for (BookieId address : addresses) {
                        BookieServiceInfo bookieServiceInfo = rc.getBookieServiceInfo(address).get().getValue();
                        compareBookieServiceInfo(bookieServiceInfo, bookieServiceInfos.get(address));
                    }
                });

    }