[fix][broker]Fix removing replica clusters on topic-policies not take effect

[AnonHxy]

Motivation

• Fix org.apache.pulsar.client.admin.Topics#removeReplicationClusters not taking effect: Because we have used HierarchyTopicPolicies to save topic-level policies, so we shoud removing policies.getReplicationClusters() != null.

Modifications

• Removing policies.getReplicationClusters() != null when topic-level policies updating
• Also, we need initialize namespace-policy first and then initialize topic-policy when PersistentTopic#initialize. The reason is that：
  • namespace-policy always has local cluster name in Policies#replication_clusters when using adminV2.
  • topic-policy may not contains local cluster name in TopicPolicies#replicationClusters.
  • If we follow the steps listed below, the persistent topic will be deleted unexpectedly, which makes the TopicPoliciesTest#testGetSetSubscribeRate test failure.
    1. admin.topics().createPartitionedTopic, which just create zookeeper nodes without PersistentTopic creating
    2. admin.namespaces().setSubscribeRate, namespace-policy will be set with local cluster name in Policies#replication_clusters. But because step.a just create zookeeper nodes, so the policy will not be applied to PersistentTopic now.
    3. admin.topicPolicies().setSubscribeRate, topic-policy will be set with no local cluster name in TopicPolicies#replicationClusters.
    4. pulsarClient1.newConsumer().subscriptionName("sub1").topic(persistenceTopic).consumerName("test").subscribe(), will create PersistentTopic
    5. if we initialize topic-policy first, the HierarchyTopicPolicies#replicationClusters will be empty , and the PersistentTopic will be deleted unexpectedly. line:1377
       

       pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

       Lines 1357 to 1378 in 603a5cd

       	public CompletableFuture<Void> checkReplication() {
       	TopicName name = TopicName.get(topic);
       	if (!name.isGlobal() || NamespaceService.isHeartbeatNamespace(name)) {
       	return CompletableFuture.completedFuture(null);
       	}
       	if (log.isDebugEnabled()) {
       	log.debug("[{}] Checking replication status", name);
       	}
       	List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
       	int newMessageTTLinSeconds = topicPolicies.getMessageTTLInSeconds().get();
       	String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
       	// if local cluster is removed from global namespace cluster-list : then delete topic forcefully
       	// because pulsar doesn't serve global topic without local repl-cluster configured.
       	if (TopicName.get(topic).isGlobal() && !configuredClusters.contains(localCluster)) {
       	log.info("Deleting topic [{}] because local cluster is not part of "
       	+ " global namespace repl list {}", topic, configuredClusters);
       	return deleteForcefully();
       	}

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

(example:)

• *Added UT org.apache.pulsar.broker.service.ReplicatorTopicPoliciesTest#testRemoveReplicationClusters

Documentation

• doc-not-needed

[Jason918]

good catch

[Jason918]

/pulsarbot run-failure-checks

[AnonHxy]

Also modify the namespace-policy and topic-policy initialization order. PTAL @Jason918

[AnonHxy]

/pulsarbot run-failure-checks

[AnonHxy]

@Technoboy- @codelipenghui PTAL

[AnonHxy]

@gaozhangmin PTAL