@Override
    public Map<String, String> findBundlesToSplit(final LoadData loadData, final PulsarService pulsar) {
        bundleCache.clear();
        namespaceBundleCount.clear();
        final ServiceConfiguration conf = pulsar.getConfiguration();
        int maxBundleCount = conf.getLoadBalancerNamespaceMaximumBundles();
        long maxBundleTopics = conf.getLoadBalancerNamespaceBundleMaxTopics();
        long maxBundleSessions = conf.getLoadBalancerNamespaceBundleMaxSessions();
        long maxBundleMsgRate = conf.getLoadBalancerNamespaceBundleMaxMsgRate();
        long maxBundleBandwidth = conf.getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * LoadManagerShared.MIBI;

        loadData.getBrokerData().forEach((broker, brokerData) -> {
            LocalBrokerData localData = brokerData.getLocalData();
            for (final Map.Entry<String, NamespaceBundleStats> entry : localData.getLastStats().entrySet()) {
                final String bundle = entry.getKey();
                final NamespaceBundleStats stats = entry.getValue();
                if (stats.topics < 2) {
                    if (log.isDebugEnabled()) {
                        log.debug("The count of topics on the bundle {} is less than 2, skip split!", bundle);
                    }
                    continue;
                }
                double totalMessageRate = 0;
                double totalMessageThroughput = 0;
                // Attempt to consider long-term message data, otherwise effectively ignore.
                if (loadData.getBundleData().containsKey(bundle)) {
                    final TimeAverageMessageData longTermData = loadData.getBundleData().get(bundle).getLongTermData();
                    totalMessageRate = longTermData.totalMsgRate();
                    totalMessageThroughput = longTermData.totalMsgThroughput();
                }
                if (stats.topics > maxBundleTopics || (maxBundleSessions > 0 && (stats.consumerCount
                        + stats.producerCount > maxBundleSessions))
                        || totalMessageRate > maxBundleMsgRate || totalMessageThroughput > maxBundleBandwidth) {
                    final String namespace = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
                    try {
                        final int bundleCount = pulsar.getNamespaceService()
                                .getBundleCount(NamespaceName.get(namespace));
                        if ((bundleCount + namespaceBundleCount.getOrDefault(namespace, 0))
                                < maxBundleCount) {
                            log.info("The bundle {} is considered to be unload. Topics: {}/{}, Sessions: ({}+{})/{}, "
                                            + "Message Rate: {}/{} (msgs/s), Message Throughput: {}/{} (MB/s)",
                                    bundle, stats.topics, maxBundleTopics, stats.producerCount, stats.consumerCount,
                                    maxBundleSessions, totalMessageRate, maxBundleMsgRate,
                                    totalMessageThroughput / LoadManagerShared.MIBI,
                                    maxBundleBandwidth / LoadManagerShared.MIBI);
                            bundleCache.put(bundle, broker);
                            int bundleNum = namespaceBundleCount.getOrDefault(namespace, 0);
                            namespaceBundleCount.put(namespace, bundleNum + 1);
                        } else {
                            if (log.isDebugEnabled()) {
                                log.debug(
                                        "Could not split namespace bundle {} because namespace {} has too many bundles:"
                                                + "{}", bundle, namespace, bundleCount);
                            }
                        }
                    } catch (Exception e) {
                        log.warn("Error while getting bundle count for namespace {}", namespace, e);
                    }
                }
            }
        });
        return bundleCache;
    }
@Test
    public void testSplitTaskWhenTopicJustOne() {
        final BundleSplitterTask bundleSplitterTask = new BundleSplitterTask();
        LoadData loadData = new LoadData();

        LocalBrokerData brokerData = new LocalBrokerData();
        Map<String, NamespaceBundleStats> lastStats = new HashMap<>();
        final NamespaceBundleStats namespaceBundleStats = new NamespaceBundleStats();
        namespaceBundleStats.topics = 1;
        lastStats.put("ten/ns/0x00000000_0x80000000", namespaceBundleStats);
        brokerData.setLastStats(lastStats);
        loadData.getBrokerData().put("broker", new BrokerData(brokerData));

        BundleData bundleData = new BundleData();
        TimeAverageMessageData averageMessageData = new TimeAverageMessageData();
        averageMessageData.setMsgRateIn(pulsar.getConfiguration().getLoadBalancerNamespaceBundleMaxMsgRate());
        averageMessageData.setMsgRateOut(1);
        bundleData.setLongTermData(averageMessageData);
        loadData.getBundleData().put("ten/ns/0x00000000_0x80000000", bundleData);

        final Map<String, String> bundlesToSplit = bundleSplitterTask.findBundlesToSplit(loadData, pulsar);
        Assert.assertEquals(bundlesToSplit.size(), 0);
    }