public CompletableFuture<Optional<LookupResult>> findRedirectLookupResultAsync() {
        String currentLMClassName = pulsar.getConfiguration().getLoadManagerClassName();
        boolean debug = ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log);
        return getAvailableBrokerLookupDataAsync().thenApply(lookupDataMap -> {
            if (lookupDataMap.isEmpty()) {
                String errorMsg = "No available broker found.";
                log.warn(errorMsg);
                throw new IllegalStateException(errorMsg);
            }
            AtomicReference<ServiceLookupData> latestServiceLookupData = new AtomicReference<>();
            AtomicLong lastStartTimestamp = new AtomicLong(0L);
            lookupDataMap.forEach((key, value) -> {
                if (lastStartTimestamp.get() <= value.getStartTimestamp()) {
                    lastStartTimestamp.set(value.getStartTimestamp());
                    latestServiceLookupData.set(value);
                }
            });
            if (latestServiceLookupData.get() == null) {
                String errorMsg = "No latest service lookup data found.";
                log.warn(errorMsg);
                throw new IllegalStateException(errorMsg);
            }

            if (Objects.equals(latestServiceLookupData.get().getLoadManagerClassName(), currentLMClassName)) {
                if (debug) {
                    log.info("No need to redirect, current load manager class name: {}",
                            currentLMClassName);
                }
                return Optional.empty();
            }
            var serviceLookupDataObj = latestServiceLookupData.get();
            var candidateBrokers = new ArrayList<ServiceLookupData>();
            lookupDataMap.forEach((key, value) -> {
                if (Objects.equals(value.getLoadManagerClassName(), serviceLookupDataObj.getLoadManagerClassName())) {
                    candidateBrokers.add(value);
                }
            });
            var selectedBroker = candidateBrokers.get((int) (Math.random() * candidateBrokers.size()));

            return Optional.of(new LookupResult(selectedBroker.getWebServiceUrl(),
                    selectedBroker.getWebServiceUrlTls(),
                    selectedBroker.getPulsarServiceUrl(),
                    selectedBroker.getPulsarServiceUrlTls(),
                    true));
        });
    }
@Test
    public void testFindRedirectLookupResultAsync() throws ExecutionException, InterruptedException {
        PulsarService pulsar = mock(PulsarService.class);
        ServiceConfiguration configuration = new ServiceConfiguration();
        when(pulsar.getConfiguration()).thenReturn(configuration);
        RedirectManager redirectManager = spy(new RedirectManager(pulsar, null));

        configuration.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
        configuration.setLoadBalancerDebugModeEnabled(true);

        // Test 1: No load manager class name found.
        doReturn(CompletableFuture.completedFuture(
                new HashMap<>(){{
                    put("broker-1", getLookupData("broker-1", null, 10));
                    put("broker-2", getLookupData("broker-2", ModularLoadManagerImpl.class.getName(), 1));
                }}
        )).when(redirectManager).getAvailableBrokerLookupDataAsync();

        // Should redirect to broker-1, since broker-1 has the latest load manager, even though the class name is null.
        Optional<LookupResult> lookupResult = redirectManager.findRedirectLookupResultAsync().get();
        assertTrue(lookupResult.isPresent());
        assertTrue(lookupResult.get().getLookupData().getBrokerUrl().contains("broker-1"));

        // Test 2: Should redirect to broker-1, since the latest broker are using ExtensibleLoadManagerImpl
        doReturn(CompletableFuture.completedFuture(
                new HashMap<>(){{
                    put("broker-1", getLookupData("broker-1", ExtensibleLoadManagerImpl.class.getName(), 10));
                    put("broker-2", getLookupData("broker-2", ModularLoadManagerImpl.class.getName(), 1));
                }}
        )).when(redirectManager).getAvailableBrokerLookupDataAsync();

        lookupResult = redirectManager.findRedirectLookupResultAsync().get();
        assertTrue(lookupResult.isPresent());
        assertTrue(lookupResult.get().getLookupData().getBrokerUrl().contains("broker-1"));


        // Test 3: Should not redirect, since current broker are using ModularLoadManagerImpl
        doReturn(CompletableFuture.completedFuture(
                new HashMap<>(){{
                    put("broker-1", getLookupData("broker-1", ExtensibleLoadManagerImpl.class.getName(), 10));
                    put("broker-2", getLookupData("broker-2", ModularLoadManagerImpl.class.getName(), 100));
                }}
        )).when(redirectManager).getAvailableBrokerLookupDataAsync();

        lookupResult = redirectManager.findRedirectLookupResultAsync().get();
        assertFalse(lookupResult.isPresent());
    }