Fix system topic replicate issue

[Technoboy-]

Motivation

PIP 92 has introduced topic policies across clusters. But after #12517, if the policy is not global, it set the replicate cluster to an empty set.

PulsarEvent.PulsarEventBuilder builder = PulsarEvent.builder();
 if (policies == null || !policies.isGlobalPolicies()) {
     // we don't need to replicate local policies to remote cluster, so set `replicateTo` to empty.
     builder.replicateTo(new HashSet<>());
}

It should set the replicateTo with the local cluster, not an empty set.

Otherwise, it will cause the system event to be replicated. Details are here :

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

Lines 319 to 328 in d4c2e61

	if (msg.hasReplicateTo() && !msg.getReplicateTo().contains(remoteCluster)) {
	if (log.isDebugEnabled()) {
	log.debug("[{}][{} -> {}] Skipping message at position {}, replicateTo {}", topicName,
	localCluster, remoteCluster, entry.getPosition(), msg.getReplicateTo());
	}
	cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
	entry.release();
	msg.recycle();
	continue;
	}

Documentation

• no-need-doc

[Jason918]

LGTM

[hangc0276]

Good job!