[Issue 8214][pulsar-client] Added pausing of new topic consumers to MultiTopicsConsumerImpl.

[sandrzejczak]

Fixes #8214

Motivation

A new consumer can be added to the MultiTopicsConsumerImpl in one of two ways:

• By invoking the subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) method manually.
• On topic partition number change (relevant method: subscribeIncreasedTopicPartitions(String topicName)).

Currently, only the second way ensures that the newly added consumers are paused if the parent consumer is paused. The first way will never result in pausing the newly added consumers.

Moreover, the unit test for the second way is not strict enough. If we modify the subscribeIncreasedTopicPartitions(String topicName) method not to pause the newly created consumers, the unit test still passes. Unit test method: SimpleProducerConsumerTest#testMultiTopicsConsumerImplPause().

Modifications

• Added pausing of the newly created consumers to the flow of the subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) method. The logic is the same as the existing logic in the subscribeIncreasedTopicPartitions(String topicName) method.
• Fixed the SimpleProducerConsumerTest#testMultiTopicsConsumerImplPause() test method, as it was too lenient. It has been renamed to SimpleProducerConsumerTest#testMultiTopicsConsumerImplPauseForPartitionNumberChange()

Verifying this change

This change added tests and can be verified as follows:

• Added the SimpleProducerConsumerTest#testMultiTopicsConsumerImplPauseForManualSubscription() unit test method.

Does this pull request potentially affect one of the following parts:

No.

Documentation

• Does this pull request introduce a new feature? no
  - If yes, how is the feature documented? not applicable
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[eolivelli]

This patch does not apply to branch-2.7