[improve][client] Enhance dynamic topic subscription management in PatternMultiTopicsConsumer

[Denovo1998]

Main Issue: #xyz

Motivation

When subscribing to topics using regex patterns, it’s possible to encounter scenarios where some topics process messages slower than others. This can lead to significant performance bottlenecks, as slower topics might block others. There’s a need to manage such situations dynamically to ensure efficient processing and fair resource allocation among topics.

Related discussions can be found in this Reddit thread, which highlights issues similar to what we aim to address.

When using MultiTopicsConsumerImpl, we can call this.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

Lines 1246 to 1258 in 5a3a1f1

	// un-subscribe a given topic
	public CompletableFuture<Void> unsubscribeAsync(String topicName) {
	checkArgument(TopicName.isValid(topicName), "Invalid topic name:" + topicName);
	if (getState() == State.Closing || getState() == State.Closed) {
	return FutureUtil.failedFuture(
	new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed"));
	}
	if (partitionsAutoUpdateTimeout != null) {
	partitionsAutoUpdateTimeout.cancel();
	partitionsAutoUpdateTimeout = null;
	}

There is a test case that explains how to unsubscribe from a single topic.

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java

Lines 660 to 663 in 5a3a1f1

	// 4, unsubscribe topic3
	CompletableFuture<Void> unsubFuture = ((MultiTopicsConsumerImpl<byte[]>) consumer).unsubscribeAsync(topicName3);
	unsubFuture.get();

This proposal seeks to integrate similar functionality into PatternMultiTopicsConsumerImpl, allowing dynamically remove or add specified topics.

Modifications

1. UnBlockTopics and BlockTopics Methods: In PatternMultiTopicsConsumerImpl, we introduce two methods: unBlockTopics and blockTopics.

• BlockTopics: Unsubscribe from topics by invoking onTopicsRemoved from TopicsChangedListener.

• UnBlockTopics: Resubscribe to topics by invoking onTopicsAdded from the same listener.

These changes will enable dynamic topic management, greatly improving the consumer’s ability to adapt to changing topic loads and enhancing overall system resilience against individual slow topics.

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[Denovo1998]

@lhotari @liangyepianzhou @poorbarcode @dao-jun PTAL!

[lhotari]

This PR appears to change the Pulsar client public API, even though the methods are protected. It remains unclear what problem this change is trying to solve. Instead of committing to a specific solution, it would be better to start a discussion on the [email protected] mailing list and reach consensus within the Pulsar project. After such a discussion, this would likely lead to creating a PIP (Pulsar Improvement Proposal).

[lhotari]

@Denovo1998 I have shared details of how to use LLMs for improving PR titles and descriptions. That would help also for this PR since the title and description is confusing.
#23806 (comment)

[Denovo1998]

@lhotari Thank you, I've learned a little trick again. Technology changes life!

And I have just started a discussion on the mailing list.