@Override
    public void filter(Set<String> brokers, BundleData bundleToAssign,
                       LoadData loadData,
                       ServiceConfiguration conf) throws BrokerFilterException {
        loadData.getBrokerData().forEach((key, value) -> {
            // The load manager class name can be null if the cluster has old version of broker.
            if (!Objects.equals(value.getLocalData().getLoadManagerClassName(), conf.getLoadManagerClassName())) {
                brokers.remove(key);
            }
        });
    }
@Test
    public void test() throws BrokerFilterException {
        BrokerLoadManagerClassFilter filter = new BrokerLoadManagerClassFilter();

        LoadData loadData = new LoadData();
        LocalBrokerData localBrokerData = new LocalBrokerData();
        localBrokerData.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());

        LocalBrokerData localBrokerData1 = new LocalBrokerData();
        localBrokerData1.setLoadManagerClassName(ExtensibleLoadManagerImpl.class.getName());

        LocalBrokerData localBrokerData2 = new LocalBrokerData();
        localBrokerData2.setLoadManagerClassName(null);
        loadData.getBrokerData().put("broker1", new BrokerData(localBrokerData));
        loadData.getBrokerData().put("broker2", new BrokerData(localBrokerData1));
        loadData.getBrokerData().put("broker3", new BrokerData(localBrokerData2));

        ServiceConfiguration conf = new ServiceConfiguration();
        conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());

        Set<String> brokers = new HashSet<>(){{
            add("broker1");
            add("broker2");
            add("broker3");
        }};
        filter.filter(brokers, null, loadData, conf);

        assertEquals(brokers.size(), 1);
        assertTrue(brokers.contains("broker1"));

        brokers = new HashSet<>(){{
            add("broker1");
            add("broker2");
            add("broker3");
        }};
        conf.setLoadManagerClassName(ExtensibleLoadManagerImpl.class.getName());
        filter.filter(brokers, null, loadData, conf);
        assertEquals(brokers.size(), 1);
        assertTrue(brokers.contains("broker2"));
    }