[Issue 6437][broker] Prevent marker messages from accumulating in backlog of replicated subscription

[massakam]

This is a further modification of the previously closed PR #6592.

Fixes #6437

Motivation

In a replicated subscription with no consumers connected, the number of marker messages in the backlog will continue to increase. If at least one active consumer is connected, the marker messages will be acknowledged and deleted by the dispatcher.

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

Lines 88 to 100 in 2d2c63e

	if (msgMetadata == null || Markers.isServerOnlyMarker(msgMetadata)) {
	PositionImpl pos = (PositionImpl) entry.getPosition();
	// Message metadata was corrupted or the messages was a server-only marker
	if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
	processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
	}
	entries.set(i, null);
	entry.release();
	subscription.acknowledgeMessage(Collections.singletonList(pos), AckType.Individual,
	Collections.emptyMap());
	continue;

On the other hand, if no consumers are connected, or if they are connected but cannot receive any messages, the dispatcher does not exist or has stopped reading entries. As a result, the marker messages accumulate in the backlog without being acknoledged by anyone.

Modifications

There are four types of marker messages:

• ReplicatedSubscriptionsSnapshotRequest
• ReplicatedSubscriptionsSnapshotResponse
• ReplicatedSubscriptionsSnapshot
• ReplicatedSubscriptionsUpdate

Of these, three messages, except ReplicatedSubscriptionsSnapshot, are not used in the local cluster. They are published to local topics for sending to remote clusters. So, modified the ReplicatedSubscriptionsController class to acknowledge these marker messages on all subscriptions immediately after publishing the messages to a local topic.

On the other hand, ReplicatedSubscriptionsSnapshot is only used by dispatchers in the local cluster and does not need to be sent to remote clusters. So, ReplicatedSubscriptionsController only acknowledges those messages on behalf of dispatcher if dispatcher has not been initialized or if there are no available consumers.

In addition, marker messages sent from remote clusters are now acknowledged by the replicator on all subscriptions.

With these changes, marker messages no longer continue to accumulate in the replicated subscription backlog.

[codelipenghui]

I'm a bit worried here. If the producer continues to produce messages but does not have consumers to consume messages, the individual ack the snapshot message will introduce more discontinuous acknowledge ranges. If that is true, the individualAcks in the cursor will get bigger and bigger.

[codelipenghui]

And if all consumers of the subscription are disconnected. The subscription also removes from this topic. So maybe can't ack the snapshot message at this moment.

[massakam]

@codelipenghui

I'm a bit worried here. If the producer continues to produce messages but does not have consumers to consume messages, the individual ack the snapshot message will introduce more discontinuous acknowledge ranges. If that is true, the individualAcks in the cursor will get bigger and bigger.

I see. Uhm...
Is there any other solution to this issue?

And if all consumers of the subscription are disconnected. The subscription also removes from this topic. So maybe can't ack the snapshot message at this moment.

If there are no subscriptions on the topic, we don't need to ack since the messages will not accumulate in the backlog, do we?

[codelipenghui]

Is there any other solution to this issue?

I also don’t have a very good way, it looks okay, just an extreme case

If there are no subscriptions on the topic, we don't need to ack since the messages will not accumulate in the backlog, do we?

make sense

[codelipenghui]

@massakam Could you please rebase to the apache/master branch? Thanks.
@merlimat @sijie @jiazhai Could you please also help review this PR?

[codelipenghui]

move to milestone 2.8.0 first