[fix][broker] Allow empty replication clusters for namespace policies

[omarkj]

This change removes validation that prevented setting empty replication clusters for namespace policies. Previously, the API would reject empty cluster lists, but this is a valid configuration for disabling inter-cluster replication while maintaining topic accessibility within individual clusters.

Changes:

• Remove CollectionUtils.isEmpty(clusterIds) validation in NamespacesBase.java
• Remove empty cluster validation in PersistentTopicsBase.java for topic-level policies
• Add comprehensive test coverage for empty replication clusters in V1 and V2 namespace APIs
• Verify bidirectional functionality (empty ↔ non-empty cluster configurations)
• Ensure no auto-population occurs when explicitly setting empty clusters

Fixes #24640

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[lhotari]

The changes so far LGTM. I'm just concerned how to remove replication clusters on the command line with pulsar-admin topics set-replication-clusters -c "" topic_name and whether the current command accepts an empty string for the -c/--clusters= parameter. Adding tests for that would be useful. However, I'm not sure if the current way of parsing the command line in the unit tests would behave in the same way as running the tool in a command shell and passing an empty argument such as -c "".

Current tests:

pulsar/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

Lines 2146 to 2147 in 34f8657

	cmdTopics.run(split("set-replication-clusters persistent://myprop/clust/ns1/ds1 -c test"));
	verify(mockTopics).setReplicationClusters("persistent://myprop/clust/ns1/ds1", Lists.newArrayList("test"));

btw. I now noticed that there's a way to remove the topic policy level configuration for replication clusters.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java

Lines 2423 to 2448 in 34f8657

	@DELETE
	@Path("/{tenant}/{namespace}/{topic}/replication")
	@ApiOperation(value = "Remove the replication clusters from a topic.")
	@ApiResponses(value = {
	@ApiResponse(code = 204, message = "Operation successful"),
	@ApiResponse(code = 403, message = "Don't have admin permission"),
	@ApiResponse(code = 404, message = "Namespace or topic doesn't exist"),
	@ApiResponse(code = 405,
	message = "Topic level policy is disabled, to enable the topic level policy and retry"),
	@ApiResponse(code = 409, message = "Concurrent modification")})
	public void removeReplicationClusters(@Suspended final AsyncResponse asyncResponse,
	@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
	@PathParam("topic") @Encoded String encodedTopic,
	@QueryParam("isGlobal") @DefaultValue("false") boolean isGlobal,
	@ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
	@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
	validateTopicName(tenant, namespace, encodedTopic);
	validateTopicPolicyOperationAsync(topicName, PolicyName.REPLICATION, PolicyOperation.WRITE)
	.thenCompose(__ -> preValidation(authoritative))
	.thenCompose(__ -> internalRemoveReplicationClusters(isGlobal))
	.thenRun(() -> asyncResponse.resume(Response.noContent().build()))
	.exceptionally(ex -> {
	handleTopicPolicyException("removeReplicationClusters", ex, asyncResponse);
	return null;
	});
	}

It maps to the pulsar-admin topics remove-replication-clusters command:

pulsar/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

Line 266 in 34f8657

addCommand("remove-replication-clusters", new RemoveReplicationClusters());

However, this isn't the same as setting the list of replication clusters to an empty list. When the topic level replication clusters is removed, it would use the namespace level setting. Being able to set it to an empty list would be necessary to override the namespace level setting so that the topic wouldn't get replicated when using a global configuration store.

[omarkj]

I would think that since there is already precedence for remove-replication-clusters within CmdTopics.java, maintaining that pattern for CmdNamespaces.java would seem reasonable just to keep those similar APIs consistent.

Your point on the subtle difference between remove-replication-clusters (reverting to the namespace setting) or setting it to the empty list (disabling it) does highlight a nuance I had not considered.

Another way to address this change would be to add an API similar to remove-replication-clusters to the Namespace API, and again disallow passing in the empty cluster set. It would then behave the same way as the topic APIs do today.

This would work for my use case. On the other hand, I do not see why the API should not allow users to disable replication for a specific topic in a replicated namespace.

[lhotari]

It seems that PIP-422 https://github.com/apache/pulsar/blob/master/pip/pip-422.md might be conflicting with the result that we'd like to achieve. PIP-422 contains behavior that once the replication clusters is updated for a global topic policy, it will delete the topics in cluster that are no longer part of the replication clusters. Automatically deleting topics in remote clusters could be considered as risky behavior and there should be a way to control this behavior since there might be a requirement where someone would simply want to later on disconnect the replication for a specific topic and continue with the topics independently. @poorbarcode Have you thought about this use case?

[poorbarcode]

Seems this PR is not needed, I have answered here