public CompletableFuture<LookupTopicResult> getBroker(TopicName topicName) {
        long startTime = System.nanoTime();
        final MutableObject<CompletableFuture> newFutureCreated = new MutableObject<>();
        try {
            return lookupInProgress.computeIfAbsent(topicName, tpName -> {
                CompletableFuture<LookupTopicResult> newFuture =
                        findBroker(serviceNameResolver.resolveHost(), false, topicName, 0);
                newFutureCreated.setValue(newFuture);

                newFuture.thenRun(() -> {
                    histoGetBroker.recordSuccess(System.nanoTime() - startTime);
                }).exceptionally(x -> {
                    histoGetBroker.recordFailure(System.nanoTime() - startTime);
                    return null;
                });
                return newFuture;
            });
        } finally {
            if (newFutureCreated.getValue() != null) {
                newFutureCreated.getValue().whenComplete((v, ex) -> {
                    lookupInProgress.remove(topicName, newFutureCreated.getValue());
                });
            }
        }
    }
@Test(invocationTimeOut = 3000)
    public void maxLookupRedirectsTest1() throws Exception {
        LookupTopicResult lookupResult = lookup.getBroker(topicName).get();
        assertEquals(lookupResult.getLogicalAddress(), InetSocketAddress
                .createUnresolved("broker2.pulsar.apache.org" ,6650));
        assertEquals(lookupResult.getPhysicalAddress(), InetSocketAddress
                .createUnresolved("broker2.pulsar.apache.org" ,6650));
        assertEquals(lookupResult.isUseProxy(), false);
    }