[fix][broker] Align bundles logic in getNamespacePoliciesAsync with sync method

Fixes #25128

Author: zooo-code

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -341,31 +341,33 @@ protected Policies getNamespacePolicies(NamespaceName namespaceName) {
     }
 
     protected CompletableFuture<Policies> getNamespacePoliciesAsync(NamespaceName namespaceName) {
-        CompletableFuture<Policies> result = new CompletableFuture<>();
-        namespaceResources().getPoliciesAsync(namespaceName)
-                .thenCombine(getLocalPolicies().getLocalPoliciesAsync(namespaceName), (pl, localPolicies) -> {
-                    if (pl.isPresent()) {
-                        Policies policies = pl.get();
-                        if (localPolicies.isPresent()) {
-                            policies.bundles = localPolicies.get().bundles;
-                            policies.migrated = localPolicies.get().migrated;
-                        }
+        return namespaceResources().getPoliciesAsync(namespaceName)
+                .thenCompose(pl -> {
+                    if (pl.isEmpty()) {
+                        return FutureUtil.failedFuture(
+                                new RestException(Status.NOT_FOUND, "Namespace does not exist"));
+                    }
+                    Policies policies = pl.get();
+                    // fetch bundles from NamespaceBundleFactory (aligns with sync method)
+                    CompletableFuture<BundlesData> bundlesFuture = pulsar().getNamespaceService()
+                            .getNamespaceBundleFactory()
+                            .getBundlesAsync(namespaceName)
+                            .thenApply(bundles -> bundles.getBundlesData());
+                    CompletableFuture<Optional<LocalPolicies>> localPoliciesFuture =
+                            getLocalPolicies().getLocalPoliciesAsync(namespaceName);
+
+                    return bundlesFuture.thenCombine(localPoliciesFuture, (bundleData, localPolicies) -> {
+                        policies.bundles = bundleData != null ? bundleData : policies.bundles;
+                        policies.migrated = localPolicies.isPresent() ? localPolicies.get().migrated : false;
                         if (policies.is_allow_auto_update_schema == null) {
                             // the type changed from boolean to Boolean. return
                             // broker value here for keeping compatibility.
                             policies.is_allow_auto_update_schema = pulsar().getConfig()
                                     .isAllowAutoUpdateSchemaEnabled();
                         }
-                        result.complete(policies);
-                    } else {
-                        result.completeExceptionally(new RestException(Status.NOT_FOUND, "Namespace does not exist"));
-                    }
-                    return null;
-                }).exceptionally(ex -> {
-                    result.completeExceptionally(ex.getCause());
-                    return null;
+                        return policies;
+                    });
                 });
-        return result;
     }
 
     protected BacklogQuota namespaceBacklogQuota(NamespaceName namespace,

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java

@@ -3429,6 +3429,36 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception {
         deleteNamespaceWithRetry(ns, false);
     }
 
+    @Test
+    public void testGetNamespacePoliciesSyncAsyncBundlesConsistency() throws Exception {
+        // Test that sync (getPolicies) and async (getPoliciesAsync) methods return consistent bundles data.
+        String ns = BrokerTestUtil.newUniqueName("prop-xyz/ns");
+        int numBundles = 16;
+
+        admin.namespaces().createNamespace(ns, numBundles);
+
+        try {
+            // Get policies using sync method
+            Policies syncPolicies = admin.namespaces().getPolicies(ns);
+
+            // Get policies using async method
+            Policies asyncPolicies = admin.namespaces().getPoliciesAsync(ns).get();
+
+            // Verify bundles are consistent between sync and async
+            assertNotNull(syncPolicies.bundles, "Sync policies should have bundles");
+            assertNotNull(asyncPolicies.bundles, "Async policies should have bundles");
+            assertEquals(asyncPolicies.bundles.getNumBundles(), syncPolicies.bundles.getNumBundles(),
+                "Number of bundles should match between sync and async");
+            assertEquals(asyncPolicies.bundles.getBoundaries(), syncPolicies.bundles.getBoundaries(),
+                "Bundle boundaries should match between sync and async");
+
+            // Also verify the expected number of bundles
+            assertEquals(syncPolicies.bundles.getNumBundles(), numBundles);
+        } finally {
+            deleteNamespaceWithRetry(ns, false);
+        }
+    }
+
     @Test
     public void testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages() throws Exception {
         final String topic = "persistent://prop-xyz/ns1/testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages";