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 maxLookupRedirectsTest3() throws Exception {
        Field field = BinaryProtoLookupService.class.getDeclaredField("maxLookupRedirects");
        field.setAccessible(true);
        field.set(lookup, 1);

        try {
            lookup.getBroker(topicName).get();
            fail("should have thrown ExecutionException");
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            assertTrue(cause instanceof LookupException);
            assertEquals(cause.getMessage(), "Too many redirects: 1");
        }
    }