@Override
    public CompletableFuture<Map<String, BrokerLookupData>> filterAsync(Map<String, BrokerLookupData> brokers,
                                                                        ServiceUnitId serviceUnit,
                                                                        LoadManagerContext context) {
        int loadBalancerBrokerMaxTopics = context.brokerConfiguration().getLoadBalancerBrokerMaxTopics();
        brokers.keySet().removeIf(broker -> {
            Optional<BrokerLoadData> brokerLoadDataOpt = context.brokerLoadDataStore().get(broker);
            long topics = brokerLoadDataOpt.map(BrokerLoadData::getTopics).orElse(0L);
            // TODO: The broker load data might be delayed, so the max topic check might not accurate.
            return topics >= loadBalancerBrokerMaxTopics;
        });
        return CompletableFuture.completedFuture(brokers);
    }
@Test
    public void test() throws IllegalAccessException, BrokerFilterException, ExecutionException, InterruptedException {
        LoadManagerContext context = getContext();
        LoadDataStore<BrokerLoadData> store = context.brokerLoadDataStore();
        BrokerLoadData maxTopicLoadData = new BrokerLoadData();
        FieldUtils.writeDeclaredField(maxTopicLoadData, "topics",
                context.brokerConfiguration().getLoadBalancerBrokerMaxTopics(), true);
        BrokerLoadData exceedMaxTopicLoadData = new BrokerLoadData();
        FieldUtils.writeDeclaredField(exceedMaxTopicLoadData, "topics",
                context.brokerConfiguration().getLoadBalancerBrokerMaxTopics() * 2, true);
        store.pushAsync("broker1", maxTopicLoadData);
        store.pushAsync("broker2", new BrokerLoadData());
        store.pushAsync("broker3", exceedMaxTopicLoadData);

        BrokerMaxTopicCountFilter filter = new BrokerMaxTopicCountFilter();
        Map<String, BrokerLookupData> originalBrokers = Map.of(
                "broker1", getLookupData(),
                "broker2", getLookupData(),
                "broker3", getLookupData(),
                "broker4", getLookupData()
        );
        Map<String, BrokerLookupData> result =
                filter.filterAsync(new HashMap<>(originalBrokers), null, context).get();
        assertEquals(result, Map.of(
                "broker2", getLookupData(),
                "broker4", getLookupData()
        ));
    }