@Override
    public synchronized Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
                                                                         final ServiceConfiguration conf) {
        selectedBundlesCache.clear();
        final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0;
        final Map<String, Long> recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles();
        final double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB;

        final double avgUsage = getBrokerAvgUsage(loadData, conf.getLoadBalancerHistoryResourcePercentage(), conf);

        if (avgUsage == 0) {
            log.warn("average max resource usage is 0");
            return selectedBundlesCache;
        }

        loadData.getBrokerData().forEach((broker, brokerData) -> {
            final LocalBrokerData localData = brokerData.getLocalData();
            final double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0);
            if (currentUsage < avgUsage + threshold) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] broker is not overloaded, ignoring at this point ({})", broker,
                            localData.printResourceUsage());
                }
                return;
            }

            double percentOfTrafficToOffload =
                    currentUsage - avgUsage - threshold + ADDITIONAL_THRESHOLD_PERCENT_MARGIN;
            double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut();
            double minimumThroughputToOffload = brokerCurrentThroughput * percentOfTrafficToOffload;

            if (minimumThroughputToOffload < minThroughputThreshold) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] broker is planning to shed throughput {} MByte/s less than "
                                    + "minimumThroughputThreshold {} MByte/s, skipping bundle unload ({})",
                            broker, minimumThroughputToOffload / MB, minThroughputThreshold / MB,
                            localData.printResourceUsage());
                }
                return;
            }

            log.info(
                    "Attempting to shed load on {}, which has max resource usage above avgUsage and threshold {}%"
                            + " > {}% + {}% -- Offloading at least {} MByte/s of traffic,"
                                    + " left throughput {} MByte/s ({})",
                    broker, 100 * currentUsage, 100 * avgUsage, 100 * threshold, minimumThroughputToOffload / MB,
                    (brokerCurrentThroughput - minimumThroughputToOffload) / MB, localData.printResourceUsage());

            if (localData.getBundles().size() > 1) {
                filterAndSelectBundle(loadData, recentlyUnloadedBundles, broker, localData, minimumThroughputToOffload);
            } 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);
            }
        });
        if (selectedBundlesCache.isEmpty() && conf.isLowerBoundarySheddingEnabled()) {
            tryLowerBoundaryShedding(loadData, conf);
        }
        return selectedBundlesCache;
    }
@Test
    public void testLowerBoundarySheddingNoBrokerToOffload() {
        int numBundles = 10;
        int brokerNum = 11;
        LoadData loadData = new LoadData();
        double throughput = 80 * 1024 * 1024;
        //Load of all Brokers are 80%, and no Broker needs to offload.
        for (int i = 0; i < brokerNum; i++) {
            LocalBrokerData broker = new LocalBrokerData();
            for (int j = 0; j < numBundles; j++) {
                broker.getBundles().add("bundle-" + j);
                BundleData bundle = new BundleData();
                TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData();
                timeAverageMessageData.setMsgThroughputIn(throughput);
                timeAverageMessageData.setMsgThroughputOut(throughput);
                bundle.setShortTermData(timeAverageMessageData);
                String broker2BundleName = "broker-" + i + "-bundle-" + j;
                loadData.getBundleData().put(broker2BundleName, bundle);
                broker.getBundles().add(broker2BundleName);
            }
            broker.setBandwidthIn(new ResourceUsage(80, 100));
            broker.setBandwidthOut(new ResourceUsage(80, 100));
            broker.setMsgThroughputIn(throughput);
            broker.setMsgThroughputOut(throughput);
            loadData.getBrokerData().put("broker-" + i, new BrokerData(broker));
        }
        ThresholdShedder shedder = new ThresholdShedder();
        Multimap<String, String> bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf);
        assertTrue(bundlesToUnload.isEmpty());
        conf.setLowerBoundarySheddingEnabled(true);
        bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf);
        assertTrue(bundlesToUnload.isEmpty());
    }