@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 testFirstPositionIsOverLoad() {
        FlowOrQpsEquallyDivideBundleSplitAlgorithm algorithm = new FlowOrQpsEquallyDivideBundleSplitAlgorithm();
        int loadBalancerNamespaceBundleMaxMsgRate = 1010;
        int loadBalancerNamespaceBundleMaxBandwidthMbytes = 100;
        int flowOrQpsDifferenceThresholdPercentage = 10;
        int topicNum = 5;

        List<String> mockTopics = new ArrayList<>();
        List<Long> topicHashList = new ArrayList<>(topicNum);
        Map<Long, String> hashAndTopic = new HashMap<>();

        for (int i = 0; i < topicNum; i++) {
            String topicName = "persistent://test-tenant1/test-namespace1/test-partition-" + i;
            mockTopics.add(topicName);
            long hashValue = Hashing.crc32().hashString(topicName, UTF_8).padToLong();
            topicHashList.add(hashValue);
            hashAndTopic.put(hashValue, topicName);
        }
        Collections.sort(topicHashList);

        Map<String, TopicStatsImpl> topicStatsMap = new HashMap<>();

        long hashValue = topicHashList.get(0);
        String topicName = hashAndTopic.get(hashValue);
        TopicStatsImpl topicStats0 = new TopicStatsImpl();
        topicStats0.msgRateIn = 1000;
        topicStats0.msgThroughputIn = 1000;
        topicStats0.msgRateOut = 1000;
        topicStats0.msgThroughputOut = 1000;
        topicStatsMap.put(topicName, topicStats0);

        for (int i = 1; i < topicHashList.size(); i++) {
            hashValue = topicHashList.get(i);
            topicName = hashAndTopic.get(hashValue);
            TopicStatsImpl topicStats = new TopicStatsImpl();
            topicStats.msgRateIn = 24.5;
            topicStats.msgThroughputIn = 1000;
            topicStats.msgRateOut = 25;
            topicStats.msgThroughputOut = 1000;
            topicStatsMap.put(topicName, 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 hash = Hashing.crc32().hashString(topic, UTF_8).padToLong();
            doReturn(hash)
                    .when(mockNamespaceBundleFactory).getLongHashCode(topic);
        });
        List<Long> splitPositions = algorithm.getSplitBoundary(new FlowOrQpsEquallyDivideBundleSplitOption(mockNamespaceService, mockNamespaceBundle,
                null, topicStatsMap, loadBalancerNamespaceBundleMaxMsgRate,
                loadBalancerNamespaceBundleMaxBandwidthMbytes, flowOrQpsDifferenceThresholdPercentage)).join();

        long splitStart = topicHashList.get(0);
        long splitEnd = topicHashList.get(1);
        long splitMiddle = splitStart + (splitEnd - splitStart) / 2 + 1;
        assertTrue(splitPositions.get(0) == splitMiddle);
    }