@Override
    public CompletableFuture<Map<String, BrokerLookupData>> filterAsync(
            Map<String, BrokerLookupData> brokers,
            ServiceUnitId serviceUnit,
            LoadManagerContext context) {
        if (brokers.isEmpty()) {
            return CompletableFuture.completedFuture(brokers);
        }
        brokers.entrySet().removeIf(entry -> {
            BrokerLookupData v = entry.getValue();
            // The load manager class name can be null if the cluster has old version of broker.
            return !Objects.equals(v.getLoadManagerClassName(),
                    context.brokerConfiguration().getLoadManagerClassName());
        });
        return CompletableFuture.completedFuture(brokers);
    }
@Test
    public void test() throws BrokerFilterException, ExecutionException, InterruptedException {
        LoadManagerContext context = getContext();
        context.brokerConfiguration().setLoadManagerClassName(ExtensibleLoadManagerImpl.class.getName());
        BrokerLoadManagerClassFilter filter = new BrokerLoadManagerClassFilter();

        Map<String, BrokerLookupData> originalBrokers = Map.of(
                "broker1", getLookupData("3.0.0", ExtensibleLoadManagerImpl.class.getName()),
                "broker2", getLookupData("3.0.0", ExtensibleLoadManagerImpl.class.getName()),
                "broker3", getLookupData("3.0.0", ModularLoadManagerImpl.class.getName()),
                "broker4", getLookupData("3.0.0", ModularLoadManagerImpl.class.getName()),
                "broker5", getLookupData("3.0.0", null)
        );

        Map<String, BrokerLookupData> result = filter.filterAsync(new HashMap<>(originalBrokers), null, context).get();
        assertEquals(result, Map.of(
                "broker1", getLookupData("3.0.0", ExtensibleLoadManagerImpl.class.getName()),
                "broker2", getLookupData("3.0.0", ExtensibleLoadManagerImpl.class.getName())
        ));

        context.brokerConfiguration().setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
        result = filter.filterAsync(new HashMap<>(originalBrokers), null, context).get();

        assertEquals(result, Map.of(
                "broker3", getLookupData("3.0.0", ModularLoadManagerImpl.class.getName()),
                "broker4", getLookupData("3.0.0", ModularLoadManagerImpl.class.getName())
        ));

    }