@Override
    public CompletableFuture<Map<String, BrokerLookupData>> filterAsync(Map<String, BrokerLookupData> availableBrokers,
                                                                        ServiceUnitId serviceUnit,
                                                                        LoadManagerContext context) {
        return isolationPoliciesHelper.applyIsolationPoliciesAsync(availableBrokers, serviceUnit)
                .thenApply(brokerCandidateCache -> {
                    availableBrokers.keySet().retainAll(brokerCandidateCache);
                    return availableBrokers;
                });
    }
@Test
    public void testFilterWithNamespaceIsolationPoliciesForPrimaryAndSecondaryBrokers()
            throws IllegalAccessException, BrokerFilterException, ExecutionException, InterruptedException {
        var namespace = "my-tenant/my-ns";
        NamespaceName namespaceName = NamespaceName.get(namespace);

        var policies = mock(SimpleResourceAllocationPolicies.class);

        // 1. Namespace: primary=broker1, secondary=broker2, shared=broker3, min_limit = 1
        setIsolationPolicies(policies, namespaceName, Set.of("broker1"), Set.of("broker2"), Set.of("broker3"), 1);
        IsolationPoliciesHelper isolationPoliciesHelper = new IsolationPoliciesHelper(policies);

        BrokerIsolationPoliciesFilter filter = new BrokerIsolationPoliciesFilter(isolationPoliciesHelper);

        // a. available-brokers: broker1, broker2, broker3 => result: broker1
        Map<String, BrokerLookupData> result = filter.filterAsync(new HashMap<>(Map.of(
                "broker1:8080", getLookupData(),
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker1:8080"));

        // b. available-brokers: broker2, broker3          => result: broker2
        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker2:8080"));

        // c. available-brokers: broker3                   => result: NULL
        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertTrue(result.isEmpty());

        // 2. Namespace: primary=broker1, secondary=broker2, shared=broker3, min_limit = 2
        setIsolationPolicies(policies, namespaceName, Set.of("broker1"), Set.of("broker2"), Set.of("broker3"), 2);

        // a. available-brokers: broker1, broker2, broker3 => result: broker1, broker2
        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker1:8080", getLookupData(),
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker1:8080", "broker2:8080"));

        // b. available-brokers: broker2, broker3          => result: broker2
        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker2:8080"));

        // c. available-brokers: broker3                   => result: NULL
        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker3:8080", getLookupData())), namespaceName, getContext()).get();
        assertTrue(result.isEmpty());
    }