@Override
    public Multimap<String, String> findBundlesForUnloading(final LoadData loadData, final ServiceConfiguration conf) {
        selectedBundlesCache.clear();
        Map<String, BrokerData> brokersData = loadData.getBrokerData();
        Map<String, BundleData> loadBundleData = loadData.getBundleDataForLoadShedding();
        Map<String, Long> recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles();

        MutableObject<String> msgRateOverloadedBroker = new MutableObject<>();
        MutableObject<String> msgThroughputOverloadedBroker = new MutableObject<>();
        MutableObject<String> msgRateUnderloadedBroker = new MutableObject<>();
        MutableObject<String> msgThroughputUnderloadedBroker = new MutableObject<>();
        MutableDouble maxMsgRate = new MutableDouble(-1);
        MutableDouble maxThroughput = new MutableDouble(-1);
        MutableDouble minMsgRate = new MutableDouble(Integer.MAX_VALUE);
        MutableDouble minThroughput = new MutableDouble(Integer.MAX_VALUE);

        brokersData.forEach((broker, data) -> {
            double msgRate = data.getLocalData().getMsgRateIn() + data.getLocalData().getMsgRateOut();
            double throughputRate = data.getLocalData().getMsgThroughputIn()
                    + data.getLocalData().getMsgThroughputOut();
            if (msgRate > maxMsgRate.getValue()) {
                msgRateOverloadedBroker.setValue(broker);
                maxMsgRate.setValue(msgRate);
            }

            if (throughputRate > maxThroughput.getValue()) {
                msgThroughputOverloadedBroker.setValue(broker);
                maxThroughput.setValue(throughputRate);
            }

            if (msgRate < minMsgRate.getValue()) {
                msgRateUnderloadedBroker.setValue(broker);
                minMsgRate.setValue(msgRate);
            }

            if (throughputRate < minThroughput.getValue()) {
                msgThroughputUnderloadedBroker.setValue(broker);
                minThroughput.setValue(throughputRate);
            }
        });

        // find the difference between two brokers based on msgRate and throughout and check if the load distribution
        // discrepancy is higher than threshold. if that matches then try to unload bundle from overloaded brokers to
        // give chance of uniform load distribution.
        if (minMsgRate.getValue() <= EPS && minMsgRate.getValue() >= -EPS) {
            minMsgRate.setValue(1.0);
        }
        if (minThroughput.getValue() <= EPS && minThroughput.getValue() >= -EPS) {
            minThroughput.setValue(1.0);
        }
        double msgRateDifferencePercentage = ((maxMsgRate.getValue() - minMsgRate.getValue()) * 100)
                / (minMsgRate.getValue());
        double msgThroughputDifferenceRate = maxThroughput.getValue() / minThroughput.getValue();

        // if the threshold matches then find out how much load needs to be unloaded by considering number of msgRate
        // and throughput.
        boolean isMsgRateThresholdExceeded = conf.getLoadBalancerMsgRateDifferenceShedderThreshold() > 0
                && msgRateDifferencePercentage > conf.getLoadBalancerMsgRateDifferenceShedderThreshold();
        boolean isMsgThroughputThresholdExceeded = conf
                .getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold() > 0
                && msgThroughputDifferenceRate > conf
                .getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold();

        if (isMsgRateThresholdExceeded || isMsgThroughputThresholdExceeded) {
            MutableInt msgRateRequiredFromUnloadedBundles = new MutableInt(
                    (int) ((maxMsgRate.getValue() - minMsgRate.getValue()) * conf.getMaxUnloadPercentage()));
            MutableInt msgThroughputRequiredFromUnloadedBundles = new MutableInt(
                    (int) ((maxThroughput.getValue() - minThroughput.getValue())
                            * conf.getMaxUnloadPercentage()));
            if (isMsgRateThresholdExceeded) {
                if (log.isDebugEnabled()) {
                    log.debug("Found bundles for uniform load balancing. "
                                    + "msgRate overloaded broker: {} with msgRate: {}, "
                                    + "msgRate underloaded broker: {} with msgRate: {}",
                            msgRateOverloadedBroker.getValue(), maxMsgRate.getValue(),
                            msgRateUnderloadedBroker.getValue(), minMsgRate.getValue());
                }
                LocalBrokerData overloadedBrokerData =
                        brokersData.get(msgRateOverloadedBroker.getValue()).getLocalData();
                if (overloadedBrokerData.getBundles().size() > 1
                        && (msgRateRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessage())) {
                    // Sort bundles by msgRate, then pick the bundle which can help to reduce load uniformly with
                    // under-loaded broker
                    loadBundleData.entrySet().stream()
                            .filter(e -> overloadedBrokerData.getBundles().contains(e.getKey()))
                            .map((e) -> {
                                String bundle = e.getKey();
                                TimeAverageMessageData shortTermData = e.getValue().getShortTermData();
                                double msgRate = shortTermData.getMsgRateIn() + shortTermData.getMsgRateOut();
                                return Pair.of(bundle, msgRate);
                            }).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()))
                            .sorted((e1, e2) -> Double.compare(e2.getRight(), e1.getRight())).forEach((e) -> {
                                if (conf.getMaxUnloadBundleNumPerShedding() != -1
                                        && selectedBundlesCache.size() >= conf.getMaxUnloadBundleNumPerShedding()) {
                                    return;
                                }
                                String bundle = e.getLeft();
                                double bundleMsgRate = e.getRight();
                                if (bundleMsgRate <= (msgRateRequiredFromUnloadedBundles.getValue()
                                        + 1000/* delta */)) {
                                    log.info("Found bundle to unload with msgRate {}", bundleMsgRate);
                                    msgRateRequiredFromUnloadedBundles.add(-bundleMsgRate);
                                    selectedBundlesCache.put(msgRateOverloadedBroker.getValue(), bundle);
                                }
                            });
                }
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("Found bundles for uniform load balancing. "
                                    + "msgThroughput overloaded broker: {} with msgThroughput {}, "
                                    + "msgThroughput underloaded broker: {} with msgThroughput: {}",
                            msgThroughputOverloadedBroker.getValue(), maxThroughput.getValue(),
                            msgThroughputUnderloadedBroker.getValue(), minThroughput.getValue());
                }
                LocalBrokerData overloadedBrokerData =
                        brokersData.get(msgThroughputOverloadedBroker.getValue()).getLocalData();
                if (overloadedBrokerData.getBundles().size() > 1
                        &&
                        msgThroughputRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessageThroughput()) {
                    // Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
                    // under-loaded broker
                    loadBundleData.entrySet().stream()
                            .filter(e -> overloadedBrokerData.getBundles().contains(e.getKey()))
                            .map((e) -> {
                                String bundle = e.getKey();
                                TimeAverageMessageData shortTermData = e.getValue().getShortTermData();
                                double msgThroughput = shortTermData.getMsgThroughputIn()
                                        + shortTermData.getMsgThroughputOut();
                                return Pair.of(bundle, msgThroughput);
                            }).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()))
                            .sorted((e1, e2) -> Double.compare(e2.getRight(), e1.getRight())).forEach((e) -> {
                                if (conf.getMaxUnloadBundleNumPerShedding() != -1
                                        && selectedBundlesCache.size() >= conf.getMaxUnloadBundleNumPerShedding()) {
                                    return;
                                }
                                String bundle = e.getLeft();
                                double msgThroughput = e.getRight();
                                if (msgThroughput <= (msgThroughputRequiredFromUnloadedBundles.getValue()
                                        + 1000/* delta */)) {
                                    log.info("Found bundle to unload with msgThroughput {}", msgThroughput);
                                    msgThroughputRequiredFromUnloadedBundles.add(-msgThroughput);
                                    selectedBundlesCache.put(msgThroughputOverloadedBroker.getValue(), bundle);
                                }
                            });
                }

            }
        }

        return selectedBundlesCache;
    }
@Test
    public void testMaxUnloadBundleNumPerShedding(){
        conf.setMaxUnloadBundleNumPerShedding(2);
        int numBundles = 20;
        LoadData loadData = new LoadData();

        LocalBrokerData broker1 = new LocalBrokerData();
        LocalBrokerData broker2 = new LocalBrokerData();

        String broker2Name = "broker2";

        double brokerThroughput = 0;

        for (int i = 1; i <= numBundles; ++i) {
            broker1.getBundles().add("bundle-" + i);

            BundleData bundle = new BundleData();

            TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData();

            double throughput = 1 * 1024 * 1024;
            timeAverageMessageData.setMsgThroughputIn(throughput);
            timeAverageMessageData.setMsgThroughputOut(throughput);
            bundle.setShortTermData(timeAverageMessageData);
            loadData.getBundleData().put("bundle-" + i, bundle);

            brokerThroughput += throughput;
        }

        broker1.setMsgThroughputIn(brokerThroughput);
        broker1.setMsgThroughputOut(brokerThroughput);

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

        Multimap<String, String> bundlesToUnload = uniformLoadShedder.findBundlesForUnloading(loadData, conf);
        assertEquals(bundlesToUnload.size(),2);
    }