@Override
    public BundleData getBundleDataOrDefault(final String bundle) {
        BundleData bundleData = null;
        try {
            Optional<BundleData> optBundleData =
                    pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join();
            if (optBundleData.isPresent()) {
                return optBundleData.get();
            }

            Optional<ResourceQuota> optQuota = pulsarResources.getLoadBalanceResources().getQuotaResources()
                    .getQuota(bundle).join();
            if (optQuota.isPresent()) {
                ResourceQuota quota = optQuota.get();
                bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES);
                // Initialize from existing resource quotas if new API ZNodes do not exist.
                final TimeAverageMessageData shortTermData = bundleData.getShortTermData();
                final TimeAverageMessageData longTermData = bundleData.getLongTermData();

                shortTermData.setMsgRateIn(quota.getMsgRateIn());
                shortTermData.setMsgRateOut(quota.getMsgRateOut());
                shortTermData.setMsgThroughputIn(quota.getBandwidthIn());
                shortTermData.setMsgThroughputOut(quota.getBandwidthOut());

                longTermData.setMsgRateIn(quota.getMsgRateIn());
                longTermData.setMsgRateOut(quota.getMsgRateOut());
                longTermData.setMsgThroughputIn(quota.getBandwidthIn());
                longTermData.setMsgThroughputOut(quota.getBandwidthOut());

                // Assume ample history.
                shortTermData.setNumSamples(NUM_SHORT_SAMPLES);
                longTermData.setNumSamples(NUM_LONG_SAMPLES);
            }
        } catch (Exception e) {
            log.warn("Error when trying to find bundle {} on metadata store: {}", bundle, e);
        }
        if (bundleData == null) {
            bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES, defaultStats);
        }
        return bundleData;
    }
@Test(dataProvider = "isV1")
    public void testBundleDataDefaultValue(boolean isV1) throws Exception {
        final String cluster = "use";
        final String tenant = "my-tenant";
        final String namespace = "my-ns";
        NamespaceName ns = isV1 ? NamespaceName.get(tenant, cluster, namespace) : NamespaceName.get(tenant, namespace);
        admin1.clusters().createCluster(cluster, ClusterData.builder().serviceUrl(pulsar1.getWebServiceAddress()).build());
        admin1.tenants().createTenant(tenant,
                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet(cluster)));
        admin1.namespaces().createNamespace(ns.toString(), 16);

        // set resourceQuota to the first bundle range.
        BundlesData bundlesData = admin1.namespaces().getBundles(ns.toString());
        NamespaceBundle namespaceBundle = nsFactory.getBundle(ns,
                Range.range(Long.decode(bundlesData.getBoundaries().get(0)), BoundType.CLOSED, Long.decode(bundlesData.getBoundaries().get(1)),
                        BoundType.OPEN));
        ResourceQuota quota = new ResourceQuota();
        quota.setMsgRateIn(1024.1);
        quota.setMsgRateOut(1024.2);
        quota.setBandwidthIn(1024.3);
        quota.setBandwidthOut(1024.4);
        quota.setMemory(1024.0);
        admin1.resourceQuotas().setNamespaceBundleResourceQuota(ns.toString(), namespaceBundle.getBundleRange(), quota);

        ModularLoadManagerWrapper loadManagerWrapper = (ModularLoadManagerWrapper) pulsar1.getLoadManager().get();
        ModularLoadManagerImpl lm = (ModularLoadManagerImpl) loadManagerWrapper.getLoadManager();

        // get the bundleData of the first bundle range.
        // The default value of the bundleData be the same as resourceQuota because the resourceQuota is present.
        BundleData defaultBundleData = lm.getBundleDataOrDefault(namespaceBundle.toString());

        TimeAverageMessageData shortTermData = defaultBundleData.getShortTermData();
        TimeAverageMessageData longTermData = defaultBundleData.getLongTermData();
        assertEquals(shortTermData.getMsgRateIn(), 1024.1);
        assertEquals(shortTermData.getMsgRateOut(), 1024.2);
        assertEquals(shortTermData.getMsgThroughputIn(), 1024.3);
        assertEquals(shortTermData.getMsgThroughputOut(), 1024.4);

        assertEquals(longTermData.getMsgRateIn(), 1024.1);
        assertEquals(longTermData.getMsgRateOut(), 1024.2);
        assertEquals(longTermData.getMsgThroughputIn(), 1024.3);
        assertEquals(longTermData.getMsgThroughputOut(), 1024.4);
    }