Only skip split the bundles that do not exist in Metadata

Author: coderzc

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

@@ -368,6 +368,13 @@ public CompletableFuture<Set<String>> getAvailableBrokersAsync() {
         return future;
     }
 
+    public boolean checkBundleDataExistInMetadataStore(String bundle) {
+        Optional<BundleData> optBundleData =
+                    pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join();
+
+        return optBundleData.isPresent();
+    }
+
     // Attempt to local the data for the given bundle in metadata store
     // If it cannot be found, return the default bundle data.
     @Override
@@ -767,6 +774,12 @@ public void checkNamespaceBundleSplit() {
                         continue;
                     }
 
+                    if (!checkBundleDataExistInMetadataStore(bundleName)) {
+                        log.warn("Bundle {} has been removed on the metadata store, skip split this bundle ",
+                            bundleName);
+                        continue;
+                    }
+
                     // Make sure the same bundle is not selected again.
                     loadData.getBundleData().remove(bundleName);
                     localData.getLastStats().remove(bundleName);
@@ -794,7 +807,6 @@ && shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
                 }
             }
 
-            writeBrokerDataOnZooKeeper(true);
             updateBundleSplitMetrics(splitCount);
         }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java

@@ -1163,7 +1163,7 @@ public void testRepeatSplitBundle() throws Exception {
         String topicToFindBundle = topicName + 0;
         NamespaceBundle realBundle = pulsar1.getNamespaceService().getBundle(TopicName.get(topicToFindBundle));
         String bundleKey = realBundle.toString();
-        log.info("bundle={}", bundleKey);
+        log.info("Before bundle={}", bundleKey);
 
         NamespaceBundleStats stats = new NamespaceBundleStats();
         stats.msgRateIn = 100000.0;