@Override
    public void writeBundleDataOnZooKeeper() {
        updateBundleData();
        // Write the bundle data to metadata store.
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        // use synchronized to protect bundleArr.
        synchronized (bundleArr) {
            int updateBundleCount = selectTopKBundle();
            bundleArr.stream().limit(updateBundleCount).forEach(entry -> futures.add(
                    pulsarResources.getLoadBalanceResources().getBundleDataResources().updateBundleData(
                            entry.getKey(), (BundleData) entry.getValue())));
        }

        // Write the time average broker data to metadata store.
        for (Map.Entry<String, BrokerData> entry : loadData.getBrokerData().entrySet()) {
            final String broker = entry.getKey();
            final TimeAverageBrokerData data = entry.getValue().getTimeAverageData();
            futures.add(pulsarResources.getLoadBalanceResources()
                    .getBrokerTimeAverageDataResources().updateTimeAverageBrokerData(broker, data));
        }

        try {
            FutureUtil.waitForAll(futures).join();
        } catch (Exception e) {
            log.warn("Error when writing metadata data to store", e);
        }
    }
@Test
    public void testFilterBundlesWhileWritingToMetadataStore() throws Exception {
        Map<String, PulsarService> pulsarServices = new HashMap<>();
        pulsarServices.put(pulsar1.getWebServiceAddress(), pulsar1);
        pulsarServices.put(pulsar2.getWebServiceAddress(), pulsar2);
        MetadataCache<BundleData> metadataCache = pulsar1.getLocalMetadataStore().getMetadataCache(BundleData.class);
        String protocol = "http://";
        PulsarService leaderBroker = pulsarServices.get(protocol + pulsar1.getLeaderElectionService().getCurrentLeader().get().getBrokerId());
        ModularLoadManagerImpl loadManager = (ModularLoadManagerImpl) getField(
                leaderBroker.getLoadManager().get(), "loadManager");
        int topK = 1;
        leaderBroker.getConfiguration().setLoadBalancerMaxNumberOfBundlesInBundleLoadReport(topK);
        // there are two broker in cluster, so total bundle count will be topK * 2
        int exportBundleCount = topK * 2;

        // create and configure bundle-data
        final int totalBundles = 5;
        final NamespaceBundle[] bundles = LoadBalancerTestingUtils.makeBundles(
                nsFactory, "test", "test", "test", totalBundles);
        LoadData loadData = (LoadData) getField(loadManager, "loadData");
        for (int i = 0; i < totalBundles; i++) {
            final BundleData bundleData = new BundleData(10, 1000);
            final String bundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[i]);
            final TimeAverageMessageData longTermMessageData = new TimeAverageMessageData(1000);
            longTermMessageData.setMsgThroughputIn(1000 * i);
            longTermMessageData.setMsgThroughputOut(1000 * i);
            longTermMessageData.setMsgRateIn(1000 * i);
            longTermMessageData.setNumSamples(1000);
            bundleData.setLongTermData(longTermMessageData);
            loadData.getBundleData().put(bundles[i].toString(), bundleData);
            loadData.getBrokerData().get(leaderBroker.getWebServiceAddress().substring(protocol.length()))
                    .getLocalData().getLastStats().put(bundles[i].toString(), new NamespaceBundleStats());
            metadataCache.create(bundleDataPath, bundleData).join();
        }
        for (int i = 0; i < totalBundles; i++) {
            final String bundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[i]);
            assertEquals(metadataCache.getWithStats(bundleDataPath).get().get().getStat().getVersion(), 0);
        }

        // update bundle data to zk and verify
        loadManager.writeBundleDataOnZooKeeper();
        int filterBundleCount = totalBundles - exportBundleCount;
        for (int i = 0; i < filterBundleCount; i++) {
            final String bundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[i]);
            assertEquals(metadataCache.getWithStats(bundleDataPath).get().get().getStat().getVersion(), 0);
        }
        for (int i = filterBundleCount; i < totalBundles; i++) {
            final String bundleDataPath = String.format("%s/%s", BUNDLE_DATA_BASE_PATH, bundles[i]);
            assertEquals(metadataCache.getWithStats(bundleDataPath).get().get().getStat().getVersion(), 1);
        }
    }