@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 testFilterWithPersistentOrNonPersistentDisabled()
            throws IllegalAccessException, BrokerFilterException, ExecutionException, InterruptedException {
        var namespace = "my-tenant/my-ns";
        NamespaceName namespaceName = NamespaceName.get(namespace);
        NamespaceBundle namespaceBundle = mock(NamespaceBundle.class);
        doReturn(true).when(namespaceBundle).hasNonPersistentTopic();
        doReturn(namespaceName).when(namespaceBundle).getNamespaceObject();

        var policies = mock(SimpleResourceAllocationPolicies.class);
        doReturn(CompletableFuture.completedFuture(false))
                .when(policies).areIsolationPoliciesPresentAsync(eq(namespaceName));
        doReturn(true).when(policies).isSharedBroker(any());
        IsolationPoliciesHelper isolationPoliciesHelper = new IsolationPoliciesHelper(policies);

        BrokerIsolationPoliciesFilter filter = new BrokerIsolationPoliciesFilter(isolationPoliciesHelper);



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


        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker1:8080", getLookupData(true, false),
                "broker2:8080", getLookupData(true, false),
                "broker3:8080", getLookupData())), namespaceBundle, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker3:8080"));

        doReturn(false).when(namespaceBundle).hasNonPersistentTopic();

        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker1:8080", getLookupData(),
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceBundle, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker1:8080", "broker2:8080", "broker3:8080"));

        result = filter.filterAsync(new HashMap<>(Map.of(
                "broker1:8080", getLookupData(false, true),
                "broker2:8080", getLookupData(),
                "broker3:8080", getLookupData())), namespaceBundle, getContext()).get();
        assertEquals(result.keySet(), Set.of("broker2:8080", "broker3:8080"));
    }