[fix][broker] Do not write replicated snapshot marker when the topic which is not enable replication

[liangyepianzhou]

Motivation

PIP 33 introduces a new concept Replicated subscriptions. When a topic has a consumer (subscription) that enables replicated subscriptions, it will write markers into the original topic. Even if there is no replicated cluster configured for this topic, the mark will still be written. And that will make the backlog of the topic keep increasing.

---

The mark will be written in the following two ways:

1. A scheduled task writes a marker at a fixed time interval if there are new messages published.
   

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

   Lines 78 to 86 in ea1fc0f

   	public ReplicatedSubscriptionsController(PersistentTopic topic, String localCluster) {
   	this.topic = topic;
   	this.localCluster = localCluster;
   	timer = topic.getBrokerService().pulsar().getExecutor()
   	.scheduleAtFixedRate(catchingAndLoggingThrowables(this::startNewSnapshot), 0,
   	topic.getBrokerService().pulsar().getConfiguration()
   	.getReplicatedSubscriptionsSnapshotFrequencyMillis(),
   	TimeUnit.MILLISECONDS);
   	}

   
   

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

   Lines 77 to 86 in ea1fc0f

   	void start() {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] Starting new snapshot {} - Clusters: {}", controller.topic().getName(), snapshotId,
   	missingClusters);
   	}
   	startTimeMillis = clock.millis();
   	controller.writeMarker(
   	Markers.newReplicatedSubscriptionsSnapshotRequest(snapshotId, controller.localCluster()));
   	}

2. Acknowledging message will trigger a check if the first snapshot is written and the mark delete position moves, if true, It will write a marker.
   

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

   Lines 114 to 150 in ea1fc0f

   	public void localSubscriptionUpdated(String subscriptionName, ReplicatedSubscriptionsSnapshot snapshot) {
   	if (log.isDebugEnabled()) {
   	log.debug("[{}][{}] Updating subscription to snapshot {}", topic, subscriptionName,
   	snapshot.getClustersList().stream()
   	.map(cmid -> String.format("%s -> %d:%d", cmid.getCluster(),
   	cmid.getMessageId().getLedgerId(), cmid.getMessageId().getEntryId()))
   	.collect(Collectors.toList()));
   	}
   	Map<String, MarkersMessageIdData> clusterIds = new TreeMap<>();
   	for (int i = 0, size = snapshot.getClustersCount(); i < size; i++) {
   	ClusterMessageId cmid = snapshot.getClusterAt(i);
   	clusterIds.put(cmid.getCluster(), cmid.getMessageId());
   	}
   	ByteBuf subscriptionUpdate = Markers.newReplicatedSubscriptionsUpdate(subscriptionName, clusterIds);
   	writeMarker(subscriptionUpdate);
   	}
   	private void receivedSnapshotRequest(ReplicatedSubscriptionsSnapshotRequest request) {
   	// if replicator producer is already closed, restart it to send snapshot response
   	Replicator replicator = topic.getReplicators().get(request.getSourceCluster());
   	if (!replicator.isConnected()) {
   	topic.startReplProducers();
   	}
   	// Send response containing the current last written message id. The response
   	// marker we're publishing locally and then replicating will have a higher
   	// message id.
   	PositionImpl lastMsgId = (PositionImpl) topic.getLastPosition();
   	if (log.isDebugEnabled()) {
   	log.debug("[{}] Received snapshot request. Last msg id: {}", topic.getName(), lastMsgId);
   	}
   	ByteBuf marker = Markers.newReplicatedSubscriptionsSnapshotResponse(
   	request.getSnapshotId(),
   	request.getSourceCluster(),

Modifications

According to the topic policy to create or remove ReplicatedSubscriptionsController of this topic.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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:

[liangyepianzhou]

The DataProvider of isTopicPolicyEnabled will be updated after the issue is fixed.

[lhotari]

Please also review #16651. That's a long out-standing PR that fixes issues with replicated snapshot marker messages.