[Bug] Logic for setting policies.bundle is different in AdminResource#getNamespacePolicies compared to AdminResource#getNamespacePoliciesAsync

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

Pulsar master branch, reading the code

Issue Description

The logic for setting policies.bundles is slightly different in org.apache.pulsar.broker.admin.AdminResource#getNamespacePolicies compared to the logic in the asynchronous method org.apache.pulsar.broker.admin.AdminResource#getNamespacePoliciesAsync.

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

Lines 318 to 369 in e041fab

	protected Policies getNamespacePolicies(NamespaceName namespaceName) {
	try {
	Policies policies = namespaceResources().getPolicies(namespaceName)
	.orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist"));
	// fetch bundles from LocalZK-policies
	BundlesData bundleData = pulsar().getNamespaceService().getNamespaceBundleFactory()
	.getBundles(namespaceName).getBundlesData();
	Optional<LocalPolicies> localPolicies = getLocalPolicies().getLocalPolicies(namespaceName);
	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();
	}
	return policies;
	} catch (RestException re) {
	throw re;
	} catch (Exception e) {
	log.error("[{}] Failed to get namespace policies {}", clientAppId(), namespaceName, e);
	throw new RestException(e);
	}
	}
	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;
	}
	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 result;
	}

Error messages

No errors

Reproducing the issue

n/a

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[oneby-wang]

Besides this, I also found that some methods implement async and async operations using nearly the same logic separately in AdminResource and its super class PulsarWebResource.

I’m wondering why we don’t just call the async method directly in the sync method and wait for the result to return. They should produce the same result.

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

Lines 454 to 472 in d72dc04

	protected Set<String> clusters() {
	try {
	// Remove "global" cluster from returned list
	Set<String> clusters = clusterResources().list().stream()
	.filter(cluster -> !Constants.GLOBAL_CLUSTER.equals(cluster)).collect(Collectors.toSet());
	return clusters;
	} catch (Exception e) {
	throw new RestException(e);
	}
	}
	protected CompletableFuture<Set<String>> clustersAsync() {
	return clusterResources().listAsync()
	.thenApply(list ->
	list.stream()
	.filter(cluster -> !Constants.GLOBAL_CLUSTER.equals(cluster))
	.collect(Collectors.toSet())
	);
	}

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

Lines 397 to 439 in d72dc04

	protected void validatePeerClusterConflict(String clusterName, Set<String> replicationClusters) {
	try {
	ClusterData clusterData = clusterResources().getCluster(clusterName).orElseThrow(
	() -> new RestException(Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName));
	Set<String> peerClusters = clusterData.getPeerClusterNames();
	if (peerClusters != null && !peerClusters.isEmpty()) {
	Sets.SetView<String> conflictPeerClusters = Sets.intersection(peerClusters, replicationClusters);
	if (!conflictPeerClusters.isEmpty()) {
	log.warn("[{}] {}'s peer cluster can't be part of replication clusters {}", clientAppId(),
	clusterName, conflictPeerClusters);
	throw new RestException(Status.CONFLICT,
	String.format("%s's peer-clusters %s can't be part of replication-clusters %s", clusterName,
	conflictPeerClusters, replicationClusters));
	}
	}
	} catch (RestException re) {
	throw re;
	} catch (Exception e) {
	log.warn("[{}] Failed to get cluster-data for {}", clientAppId(), clusterName, e);
	}
	}
	protected CompletableFuture<Void> validatePeerClusterConflictAsync(String clusterName,
	Set<String> replicationClusters) {
	return clusterResources().getClusterAsync(clusterName)
	.thenAccept(data -> {
	ClusterData clusterData = data.orElseThrow(() -> new RestException(
	Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName));
	Set<String> peerClusters = clusterData.getPeerClusterNames();
	if (peerClusters != null && !peerClusters.isEmpty()) {
	Sets.SetView<String> conflictPeerClusters =
	Sets.intersection(peerClusters, replicationClusters);
	if (!conflictPeerClusters.isEmpty()) {
	log.warn("[{}] {}'s peer cluster can't be part of replication clusters {}", clientAppId(),
	clusterName, conflictPeerClusters);
	throw new RestException(Status.CONFLICT,
	String.format("%s's peer-clusters %s can't be part of replication-clusters %s",
	clusterName,
	conflictPeerClusters, replicationClusters));
	}
	}
	});
	}

[zooo-code]

Hi @lhotari , @oneby-wang ! I'd like to work on this issue if it's available.

I've reviewed the code and noticed a discrepancy between the two methods:

• getNamespacePolicies() (sync, deprecated):
  Uses NamespaceBundleFactory.getBundles().getBundlesData() with a fallback

• getNamespacePoliciesAsync():
  Uses LocalPolicies.bundles without the fallback logic

Before proposing a fix, I have a question:

Which source is the correct one for bundle data?

• Should we use NamespaceBundleFactory (as the sync method does)?
  Or is LocalPolicies the intended source (as the async method currently does)?

I noticed that getBundlesAsync() exists in NamespaceBundleFactory
(used in NamespacesBase.java:364), so we could potentially apply the same logic to the async method.

Also, regarding the recent PR #24762 that fixed bundle-related issues — should this change align with that approach?

Could you provide some guidance on the intended behavior? Once clarified, I'm happy to submit a PR with tests.

Thanks!!

[zooo-code]

Hi @lhotari @oneby-wang ,

Since there hasn't been any objection, I’ll proceed with creating a PR to address this discrepancy. I plan to implement the fix by aligning the getNamespacePoliciesAsync logic with the NamespaceBundleFactory approach (similar to the sync version).

I will submit the PR shortly.
I'm happy to adjust the implementation during the review if a different approach is preferred.

Thanks!!!!

[lhotari]

I’m wondering why we don’t just call the async method directly in the sync method and wait for the result to return. They should produce the same result.

@oneby-wang @zooo-code I think that this makes sense in this case. There could be some change in behavior when doing that, but it's unlikely that it causes problems in this case.

[zooo-code]

Thanks for the feedback @lhotari !

I agree that having sync call async would be cleaner and eliminate code duplication.
Should I update my current PR #25158 to refactor it that way? Or would you prefer to merge the current fix first and handle the refactoring in a separate PR?

Happy to go either way!

[lhotari]

Thanks for the feedback @lhotari !

I agree that having sync call async would be cleaner and eliminate code duplication. Should I update my current PR #25158 to refactor it that way? Or would you prefer to merge the current fix first and handle the refactoring in a separate PR?

Happy to go either way!

I think it makes sense to handle it in the current PR.

[zooo-code]

Updated PR #25158 based on @lhotari 's and @oneby-wang 's feedback:

• Refactored sync method to call async method (eliminating code duplication)
• Added test testGetNamespacePoliciesSyncAsyncBundlesConsistency to verify consistency

Please let me know if any further adjustments are needed. Thanks for the guidance!