[Broker] Fix NPEs and thread safety issue in PersistentReplicator

[lhotari]

Fixes #9762

Motivation

NPEs occurs in PersistentReplicator, see #9762 for details.

Modifications

• Make cursor field volatile since the field is updated asynchronously in another thread. -
• Remove the unnecessary synchronization on openCursorAsync method since it's not needed.
• Add null checks before accessing the cursor field since statistics might be updated before the cursor is available. Call comes from

  pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarStats.java

  Lines 94 to 133 in cc64889

  	public synchronized void updateStats(
  	ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, Topic>>>
  	topicsMap) {
  	StatsOutputStream topicStatsStream = new StatsOutputStream(tempTopicStatsBuf);
  	try {
  	tempMetricsCollection.clear();
  	bundleStats.clear();
  	brokerOperabilityMetrics.reset();
  	// Json begin
  	topicStatsStream.startObject();
  	topicsMap.forEach((namespaceName, bundles) -> {
  	if (bundles.isEmpty()) {
  	return;
  	}
  	try {
  	topicStatsStream.startObject(namespaceName);
  	nsStats.reset();
  	bundles.forEach((bundle, topics) -> {
  	NamespaceBundleStats currentBundleStats = bundleStats.computeIfAbsent(bundle,
  	k -> new NamespaceBundleStats());
  	currentBundleStats.reset();
  	currentBundleStats.topics = topics.size();
  	topicStatsStream.startObject(NamespaceBundle.getBundleRange(bundle));
  	tempNonPersistentTopics.clear();
  	// start persistent topic
  	topicStatsStream.startObject("persistent");
  	topics.forEach((name, topic) -> {
  	if (topic instanceof PersistentTopic) {
  	try {
  	topic.updateRates(nsStats, currentBundleStats, topicStatsStream,
  	clusterReplicationMetrics, namespaceName, exposePublisherStats);

[lhotari]

/pulsarbot run-failure-checks