public CompletableFuture<LookupResult> createLookupResult(String candidateBroker, boolean authoritativeRedirect,
                                                                 final String advertisedListenerName) {

        CompletableFuture<LookupResult> lookupFuture = new CompletableFuture<>();
        try {
            checkArgument(StringUtils.isNotBlank(candidateBroker), "Lookup broker can't be null %s", candidateBroker);
            String path = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + candidateBroker;

            localBrokerDataCache.get(path).thenAccept(reportData -> {
                if (reportData.isPresent()) {
                    LocalBrokerData lookupData = reportData.get();
                    if (StringUtils.isNotBlank(advertisedListenerName)) {
                        AdvertisedListener listener = lookupData.getAdvertisedListeners().get(advertisedListenerName);
                        if (listener == null) {
                            lookupFuture.completeExceptionally(
                                    new PulsarServerException(
                                            "the broker do not have " + advertisedListenerName + " listener"));
                        } else {
                            URI url = listener.getBrokerServiceUrl();
                            URI urlTls = listener.getBrokerServiceUrlTls();
                            lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
                                    lookupData.getWebServiceUrlTls(), url == null ? null : url.toString(),
                                    urlTls == null ? null : urlTls.toString(), authoritativeRedirect));
                        }
                    } else {
                        lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
                                lookupData.getWebServiceUrlTls(), lookupData.getPulsarServiceUrl(),
                                lookupData.getPulsarServiceUrlTls(), authoritativeRedirect));
                    }
                } else {
                    lookupFuture.completeExceptionally(new MetadataStoreException.NotFoundException(path));
                }
            }).exceptionally(ex -> {
                lookupFuture.completeExceptionally(ex);
                return null;
            });
        } catch (Exception e) {
            lookupFuture.completeExceptionally(e);
        }
        return lookupFuture;
    }
@Test
    public void testLoadReportDeserialize() throws Exception {

        final String candidateBroker1 = "localhost:8000";
        String broker1Url = "pulsar://localhost:6650";
        final String candidateBroker2 = "localhost:3000";
        String broker2Url = "pulsar://localhost:6660";
        LoadReport lr = new LoadReport("http://" + candidateBroker1, null, broker1Url, null);
        LocalBrokerData ld = new LocalBrokerData("http://" + candidateBroker2, null, broker2Url, null);
        String path1 = String.format("%s/%s", LoadManager.LOADBALANCE_BROKERS_ROOT, candidateBroker1);
        String path2 = String.format("%s/%s", LoadManager.LOADBALANCE_BROKERS_ROOT, candidateBroker2);

        pulsar.getLocalMetadataStore().put(path1,
                ObjectMapperFactory.getMapper().writer().writeValueAsBytes(lr),
                Optional.empty(),
                EnumSet.of(CreateOption.Ephemeral)
                ).join();
        pulsar.getLocalMetadataStore().put(path2,
                ObjectMapperFactory.getMapper().writer().writeValueAsBytes(ld),
                Optional.empty(),
                EnumSet.of(CreateOption.Ephemeral)
        ).join();
        LookupResult result1 = pulsar.getNamespaceService().createLookupResult(candidateBroker1, false, null).get();

        // update to new load manager
        LoadManager oldLoadManager = pulsar.getLoadManager()
                .getAndSet(new ModularLoadManagerWrapper(new ModularLoadManagerImpl()));
        oldLoadManager.stop();
        LookupResult result2 = pulsar.getNamespaceService().createLookupResult(candidateBroker2, false, null).get();
        Assert.assertEquals(result1.getLookupData().getBrokerUrl(), broker1Url);
        Assert.assertEquals(result2.getLookupData().getBrokerUrl(), broker2Url);
        System.out.println(result2);
    }