@Override
    public CompletableFuture<Map<String, BrokerLookupData>> filterAsync(Map<String, BrokerLookupData> brokers,
                                                                        ServiceUnitId serviceUnit,
                                                                        LoadManagerContext context) {
        ServiceConfiguration conf = context.brokerConfiguration();
        if (!conf.isPreferLaterVersions() || brokers.isEmpty()) {
            return CompletableFuture.completedFuture(brokers);
        }

        Version latestVersion;
        try {
            latestVersion = getLatestVersionNumber(brokers);
            if (log.isDebugEnabled()) {
                log.debug("Latest broker version found was [{}]", latestVersion);
            }
        } catch (Exception ex) {
            log.warn("Disabling PreferLaterVersions feature; reason: " + ex.getMessage());
            return FutureUtil.failedFuture(
                    new BrokerFilterBadVersionException("Cannot determine newest broker version: " + ex.getMessage()));
        }

        int numBrokersLatestVersion = 0;
        int numBrokersOlderVersion = 0;

        Iterator<Map.Entry<String, BrokerLookupData>> brokerIterator = brokers.entrySet().iterator();
        while (brokerIterator.hasNext()) {
            Map.Entry<String, BrokerLookupData> next = brokerIterator.next();
            String brokerId = next.getKey();
            String version = next.getValue().brokerVersion();
            Version brokerVersionVersion = Version.valueOf(version);
            if (brokerVersionVersion.equals(latestVersion)) {
                log.debug("Broker [{}] is running the latest version ([{}])", brokerId, version);
                numBrokersLatestVersion++;
            } else {
                log.info("Broker [{}] is running an older version ([{}]); latest version is [{}]",
                        brokerId, version, latestVersion);
                numBrokersOlderVersion++;
                brokerIterator.remove();
            }
        }
        if (numBrokersOlderVersion == 0) {
            log.info("All {} brokers are running the latest version [{}]", numBrokersLatestVersion, latestVersion);
        }
        return CompletableFuture.completedFuture(brokers);
    }
@Test
    public void testFilterEmptyBrokerList() throws BrokerFilterException, ExecutionException, InterruptedException {
        BrokerVersionFilter brokerVersionFilter = new BrokerVersionFilter();
        Map<String, BrokerLookupData> result = brokerVersionFilter.filterAsync(new HashMap<>(), null, getContext()).get();
        assertTrue(result.isEmpty());
    }