@Override
    public CompletableFuture<List<Long>> getSplitBoundary(BundleSplitOption bundleSplitOptionTmp) {
        FlowOrQpsEquallyDivideBundleSplitOption bundleSplitOption =
                (FlowOrQpsEquallyDivideBundleSplitOption) bundleSplitOptionTmp;
        NamespaceService service = bundleSplitOption.getService();
        NamespaceBundle bundle = bundleSplitOption.getBundle();
        Map<String, TopicStatsImpl> topicStatsMap = bundleSplitOption.getTopicStatsMap();
        int loadBalancerNamespaceBundleMaxMsgRate = bundleSplitOption.getLoadBalancerNamespaceBundleMaxMsgRate();
        double diffThreshold = bundleSplitOption.getFlowOrQpsDifferenceThresholdPercentage() / 100.0;
        long loadBalancerNamespaceBundleMaxBandwidthBytes = bundleSplitOption
                .getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * MBytes;


        return service.getOwnedTopicListForNamespaceBundle(bundle).thenCompose(topics -> {
            if (topics == null || topics.size() <= 1) {
                return CompletableFuture.completedFuture(null);
            }

            double bundleThroughput = 0;
            double bundleMsgRate = 0;
            Map<Long, TopicInfo> topicInfoMap = new HashMap<>();
            List<Long> topicHashList = new ArrayList<>(topics.size());
            for (String topic : topics) {
                TopicStatsImpl topicStats = topicStatsMap.get(topic);
                if (topicStats == null) {
                    continue;
                }
                double msgRateIn = topicStats.getMsgRateIn();
                double msgRateOut = topicStats.getMsgRateOut();
                double msgThroughputIn = topicStats.getMsgThroughputIn();
                double msgThroughputOut = topicStats.getMsgThroughputOut();
                double msgRate = msgRateIn + msgRateOut;
                double throughput = msgThroughputIn + msgThroughputOut;
                if (msgRate <= 0 && throughput <= 0) {
                    // Skip empty topic
                    continue;
                }

                Long hashCode = bundle.getNamespaceBundleFactory().getLongHashCode(topic);
                topicHashList.add(hashCode);
                topicInfoMap.put(hashCode, new TopicInfo(topic, msgRate, throughput));
                bundleThroughput += throughput;
                bundleMsgRate += msgRate;
            }

            if (topicInfoMap.size() < 2
                    || (bundleMsgRate < (loadBalancerNamespaceBundleMaxMsgRate * (1 + diffThreshold))
                    && bundleThroughput < (loadBalancerNamespaceBundleMaxBandwidthBytes * (1 + diffThreshold)))) {
                return CompletableFuture.completedFuture(null);
            }
            Collections.sort(topicHashList);


            List<Long> splitResults = new ArrayList<>();
            double bundleMsgRateTmp = topicInfoMap.get(topicHashList.get(0)).msgRate;
            double bundleThroughputTmp = topicInfoMap.get(topicHashList.get(0)).throughput;

            for (int i = 1; i < topicHashList.size(); i++) {
                long topicHashCode = topicHashList.get(i);
                double msgRate = topicInfoMap.get(topicHashCode).msgRate;
                double throughput = topicInfoMap.get(topicHashCode).throughput;

                if ((bundleMsgRateTmp + msgRate) > loadBalancerNamespaceBundleMaxMsgRate
                        || (bundleThroughputTmp + throughput) > loadBalancerNamespaceBundleMaxBandwidthBytes) {
                    long splitStart = topicHashList.get(i - 1);
                    long splitEnd = topicHashList.get(i);
                    long splitMiddle = splitStart + (splitEnd - splitStart) / 2 + 1;
                    splitResults.add(splitMiddle);

                    bundleMsgRateTmp = msgRate;
                    bundleThroughputTmp = throughput;
                } else {
                    bundleMsgRateTmp += msgRate;
                    bundleThroughputTmp += throughput;
                }
            }

            return CompletableFuture.completedFuture(splitResults);
        });
    }
@Test
    public void testSplitBundleByFlowOrQps() {
        FlowOrQpsEquallyDivideBundleSplitAlgorithm algorithm = new FlowOrQpsEquallyDivideBundleSplitAlgorithm();
        int loadBalancerNamespaceBundleMaxMsgRate = 1010;
        int loadBalancerNamespaceBundleMaxBandwidthMbytes = 100;
        int flowOrQpsDifferenceThresholdPercentage = 10;

        Map<Long, Double> hashAndMsgMap = new HashMap<>();
        Map<Long, Double> hashAndThroughput = new HashMap<>();
        Map<String, TopicStatsImpl> topicStatsMap = new HashMap<>();
        List<String> mockTopics = new ArrayList<>();
        List<Long> hashList = new ArrayList<>();

        for (int i = 1; i < 6; i++) {
            String topicName = "persistent://test-tenant1/test-namespace1/test" + i;
            for (int j = 0; j < 20; j++) {
                String tp = topicName + "-partition-" + j;
                mockTopics.add(tp);
                TopicStatsImpl topicStats = new TopicStatsImpl();
                topicStats.msgRateIn = 24.5;
                topicStats.msgThroughputIn = 1000;
                topicStats.msgRateOut = 25;
                topicStats.msgThroughputOut = 1000;
                topicStatsMap.put(tp, topicStats);
            }
        }


        for (int i = 6; i < 13; i++) {
            String topicName = "persistent://test-tenant1/test-namespace1/test" + i;
            for (int j = 0; j < 20; j++) {
                String tp = topicName + "-partition-" + j;
                mockTopics.add(tp);
                TopicStatsImpl topicStats = new TopicStatsImpl();
                topicStats.msgRateIn = 25.5;
                topicStats.msgThroughputIn = 1000;
                topicStats.msgRateOut = 25;
                topicStats.msgThroughputOut = 1000;
                topicStatsMap.put(tp, topicStats);
            }
        }

        String tp = "persistent://test-tenant1/test-namespace1/test695-partition-0";
        mockTopics.add(tp);
        TopicStatsImpl topicStats = new TopicStatsImpl();
        topicStats.msgRateIn = 25;
        topicStats.msgThroughputIn = 1000;
        topicStats.msgRateOut = 35;
        topicStats.msgThroughputOut = 1000;
        topicStatsMap.put(tp, topicStats);

        // -- do test
        NamespaceService mockNamespaceService = mock(NamespaceService.class);
        NamespaceBundle mockNamespaceBundle = mock(NamespaceBundle.class);
        doReturn(CompletableFuture.completedFuture(mockTopics))
                .when(mockNamespaceService).getOwnedTopicListForNamespaceBundle(mockNamespaceBundle);
        NamespaceBundleFactory mockNamespaceBundleFactory = mock(NamespaceBundleFactory.class);
        doReturn(mockNamespaceBundleFactory)
                .when(mockNamespaceBundle).getNamespaceBundleFactory();
        mockTopics.forEach((topic) -> {
            long hashValue = Hashing.crc32().hashString(topic, UTF_8).padToLong();
            doReturn(hashValue)
                    .when(mockNamespaceBundleFactory).getLongHashCode(topic);
            hashList.add(hashValue);
            hashAndMsgMap.put(hashValue, topicStatsMap.get(topic).msgRateIn
                    + topicStatsMap.get(topic).msgRateOut);
            hashAndThroughput.put(hashValue, topicStatsMap.get(topic).msgThroughputIn
                    + topicStatsMap.get(topic).msgThroughputOut);
        });

        List<Long> splitPositions = algorithm.getSplitBoundary(new FlowOrQpsEquallyDivideBundleSplitOption(mockNamespaceService, mockNamespaceBundle,
                null, topicStatsMap, loadBalancerNamespaceBundleMaxMsgRate,
                loadBalancerNamespaceBundleMaxBandwidthMbytes, flowOrQpsDifferenceThresholdPercentage)).join();

        Collections.sort(hashList);
        int i = 0;
        for (Long position : splitPositions) {
            Long endPosition = position;
            double bundleMsgRateTmp = 0;
            double bundleThroughputTmp = 0;
            while (hashList.get(i) < endPosition) {
                bundleMsgRateTmp += hashAndMsgMap.get(hashList.get(i));
                bundleThroughputTmp += hashAndThroughput.get(hashList.get(i));
                i++;
            }
            assertTrue(bundleMsgRateTmp < loadBalancerNamespaceBundleMaxMsgRate);
            assertTrue(bundleThroughputTmp < loadBalancerNamespaceBundleMaxBandwidthMbytes * 1024 * 1024);
        }
    }