@Override
    public Multimap<String, String> findBundlesForUnloading(final LoadData loadData, final ServiceConfiguration conf) {
        selectedBundlesCache.clear();
        final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage() / 100.0;
        final Map<String, Long> recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles();

        // Check every broker and select
        loadData.getBrokerData().forEach((broker, brokerData) -> {

            final LocalBrokerData localData = brokerData.getLocalData();
            final double currentUsage = localData.getMaxResourceUsageWithWeight(
                    conf.getLoadBalancerCPUResourceWeight(),
                    conf.getLoadBalancerDirectMemoryResourceWeight(),
                    conf.getLoadBalancerBandwidthInResourceWeight(),
                    conf.getLoadBalancerBandwidthOutResourceWeight());
            if (currentUsage < overloadThreshold) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] Broker is not overloaded, ignoring at this point ({})", broker,
                            localData.printResourceUsage());
                }
                return;
            }

            // We want to offload enough traffic such that this broker will go below the overload threshold
            // Also, add a small margin so that this broker won't be very close to the threshold edge.
            double percentOfTrafficToOffload = currentUsage - overloadThreshold + ADDITIONAL_THRESHOLD_PERCENT_MARGIN;
            double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut();

            double minimumThroughputToOffload = brokerCurrentThroughput * percentOfTrafficToOffload;

            log.info(
                    "Attempting to shed load on {}, which has resource usage {}% above threshold {}%"
                            + " -- Offloading at least {} MByte/s of traffic ({})",
                    broker, 100 * currentUsage, 100 * overloadThreshold, minimumThroughputToOffload / 1024 / 1024,
                    localData.printResourceUsage());

            MutableDouble trafficMarkedToOffload = new MutableDouble(0);
            MutableBoolean atLeastOneBundleSelected = new MutableBoolean(false);

            if (localData.getBundles().size() > 1) {
                // Sort bundles by throughput, then pick the biggest N which combined
                // make up for at least the minimum throughput to offload

                loadData.getBundleDataForLoadShedding().entrySet().stream()
                    .filter(e -> localData.getBundles().contains(e.getKey()))
                    .map((e) -> {
                        // Map to throughput value
                        // Consider short-term byte rate to address system resource burden
                        String bundle = e.getKey();
                        BundleData bundleData = e.getValue();
                        TimeAverageMessageData shortTermData = bundleData.getShortTermData();
                        double throughput = shortTermData.getMsgThroughputIn() + shortTermData
                                .getMsgThroughputOut();
                    return Pair.of(bundle, throughput);
                }).filter(e -> {
                    // Only consider bundles that were not already unloaded recently
                    return !recentlyUnloadedBundles.containsKey(e.getLeft());
                }).sorted((e1, e2) -> {
                    // Sort by throughput in reverse order
                    return Double.compare(e2.getRight(), e1.getRight());
                }).forEach(e -> {
                    if (trafficMarkedToOffload.doubleValue() < minimumThroughputToOffload
                            || atLeastOneBundleSelected.isFalse()) {
                       selectedBundlesCache.put(broker, e.getLeft());
                       trafficMarkedToOffload.add(e.getRight());
                       atLeastOneBundleSelected.setTrue();
                   }
                });
            } else if (localData.getBundles().size() == 1) {
                log.warn(
                        "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
                                + "No Load Shedding will be done on this broker",
                        localData.getBundles().iterator().next(), broker);
            } else {
                log.warn("Broker {} is overloaded despite having no bundles", broker);
            }

        });

        return selectedBundlesCache;
    }
@Test
    public void testBrokerNotOverloaded() {
        LoadData loadData = new LoadData();

        LocalBrokerData broker1 = new LocalBrokerData();
        broker1.setBandwidthIn(new ResourceUsage(500, 1000));
        broker1.setBandwidthOut(new ResourceUsage(500, 1000));
        broker1.setBundles(Sets.newHashSet("bundle-1"));

        BundleData bundle1 = new BundleData();
        TimeAverageMessageData db1 = new TimeAverageMessageData();
        db1.setMsgThroughputIn(1000);
        db1.setMsgThroughputOut(1000);
        bundle1.setShortTermData(db1);
        loadData.getBundleData().put("bundle-1", bundle1);

        loadData.getBrokerData().put("broker-1", new BrokerData(broker1));

        assertTrue(os.findBundlesForUnloading(loadData, conf).isEmpty());
    }