repeat metadata and cache update

Author: heesung-sn

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -89,6 +89,7 @@
 import org.apache.pulsar.common.naming.NamespaceBundle;
 import org.apache.pulsar.common.naming.NamespaceBundleFactory;
 import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
+import org.apache.pulsar.common.naming.NamespaceBundles;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
@@ -852,8 +853,10 @@ protected void splitServiceUnitOnceAndRetry(NamespaceService namespaceService,
                                                 long startTime,
                                                 CompletableFuture<Void> completionFuture) {
         ownChildBundles(childBundles, parentData)
-                .thenCompose(__ -> updateSplitNamespaceBundlesAsync(
+                .thenCompose(__ -> getSplitNamespaceBundles(
                         namespaceService, bundleFactory, algorithm, parentBundle, childBundles, boundaries))
+                .thenCompose(namespaceBundles -> updateSplitNamespaceBundlesAsync(
+                        namespaceService, bundleFactory, parentBundle, namespaceBundles))
                 .thenAccept(__ -> // Update bundled_topic cache for load-report-generation
                         pulsar.getBrokerService().refreshTopicToStatsMaps(parentBundle))
                 .thenAccept(__ -> pubAsync(parentBundle.toString(), new ServiceUnitStateData(
@@ -913,69 +916,69 @@ private CompletableFuture<Void> ownChildBundles(List<NamespaceBundle> childBundl
         }
     }
 
-    private CompletableFuture<Void> updateSplitNamespaceBundlesAsync(
-            NamespaceService namespaceService,
-            NamespaceBundleFactory bundleFactory,
-            NamespaceBundleSplitAlgorithm algorithm,
-            NamespaceBundle parentBundle,
-            List<NamespaceBundle> childBundles,
-            List<Long> boundaries) {
-        CompletableFuture<Void> updateSplitNamespaceBundlesFuture = new CompletableFuture<>();
-        var namespaceName = parentBundle.getNamespaceObject();
+    private CompletableFuture<NamespaceBundles> getSplitNamespaceBundles(NamespaceService namespaceService,
+                                                                         NamespaceBundleFactory bundleFactory,
+                                                                         NamespaceBundleSplitAlgorithm algorithm,
+                                                                         NamespaceBundle parentBundle,
+                                                                         List<NamespaceBundle> childBundles,
+                                                                         List<Long> boundaries) {
+        CompletableFuture future = new CompletableFuture();
         final var debug = debug();
         var targetNsBundle = bundleFactory.getBundles(parentBundle.getNamespaceObject());
-        boolean updated = false;
+        boolean found = false;
         try {
             targetNsBundle.validateBundle(parentBundle);
         } catch (IllegalArgumentException e) {
             if (debug) {
                 log.info("Namespace bundles do not contain the parent bundle:{}",
                         parentBundle);
             }
-            for (var childBundle : childBundles){
+            for (var childBundle : childBundles) {
                 try {
                     targetNsBundle.validateBundle(childBundle);
                     if (debug) {
                         log.info("Namespace bundles contain the child bundle:{}",
                                 childBundle);
                     }
                 } catch (Exception ex) {
-                    updateSplitNamespaceBundlesFuture.completeExceptionally(
+                    future.completeExceptionally(
                             new BrokerServiceException.ServiceUnitNotReadyException(
                                     "Namespace bundles do not contain the child bundle:" + childBundle, e));
-                    return updateSplitNamespaceBundlesFuture;
+                    return future;
                 }
             }
-            updated = true;
+            found = true;
         } catch (Exception e) {
-            updateSplitNamespaceBundlesFuture.completeExceptionally(
+            future.completeExceptionally(
                     new BrokerServiceException.ServiceUnitNotReadyException(
                             "Failed to validate the parent bundle in the namespace bundles.", e));
-            return updateSplitNamespaceBundlesFuture;
+            return future;
         }
-        if (updated) {
-            updateSplitNamespaceBundlesFuture.complete(null);
+        if (found) {
+            future.complete(targetNsBundle);
+            return future;
         } else {
-            namespaceService.getSplitBoundary(parentBundle, algorithm, boundaries)
-                    .thenApply(splitBundlesPair -> splitBundlesPair.getLeft())
-                    .thenCompose(splitNamespaceBundles ->
-                            namespaceService.updateNamespaceBundles(
-                                            namespaceName, splitNamespaceBundles)
-                                    .thenCompose(__ -> namespaceService.updateNamespaceBundlesForPolicies(
-                                            namespaceName, splitNamespaceBundles)))
-                    .thenAccept(__ -> {
-                        bundleFactory.invalidateBundleCache(parentBundle.getNamespaceObject());
-                        if (debug) {
-                            log.info("Successfully updated split namespace bundles and namespace bundle cache.");
-                        }
-                        updateSplitNamespaceBundlesFuture.complete(null);
-                    })
-                    .exceptionally(ex -> {
-                        updateSplitNamespaceBundlesFuture.completeExceptionally(ex);
-                        return null;
-                    });
+            return namespaceService.getSplitBoundary(parentBundle, algorithm, boundaries)
+                    .thenApply(splitBundlesPair -> splitBundlesPair.getLeft());
         }
-        return updateSplitNamespaceBundlesFuture;
+    }
+
+    private CompletableFuture<Void> updateSplitNamespaceBundlesAsync(
+            NamespaceService namespaceService,
+            NamespaceBundleFactory bundleFactory,
+            NamespaceBundle parentBundle,
+            NamespaceBundles splitNamespaceBundles) {
+        var namespaceName = parentBundle.getNamespaceObject();
+        return namespaceService.updateNamespaceBundles(
+                        namespaceName, splitNamespaceBundles)
+                .thenCompose(__ -> namespaceService.updateNamespaceBundlesForPolicies(
+                        namespaceName, splitNamespaceBundles))
+                .thenAccept(__ -> {
+                    bundleFactory.invalidateBundleCache(parentBundle.getNamespaceObject());
+                    if (debug()) {
+                        log.info("Successfully updated split namespace bundles and namespace bundle cache.");
+                    }
+                });
     }
 
     public void handleMetadataSessionEvent(SessionEvent e) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java

@@ -1316,7 +1316,7 @@ public void splitAndRetryFailureTest() throws Exception {
             doReturn(CompletableFuture.completedFuture(List.of("test-topic-1", "test-topic-2")))
                     .when(namespaceService).getOwnedTopicListForNamespaceBundle(any());
             return future;
-        }).when(namespaceService).updateNamespaceBundles(any(), any());
+        }).when(namespaceService).updateNamespaceBundlesForPolicies(any(), any());
         doReturn(namespaceService).when(pulsar1).getNamespaceService();
         doReturn(CompletableFuture.completedFuture(List.of("test-topic-1", "test-topic-2")))
                 .when(namespaceService).getOwnedTopicListForNamespaceBundle(any());