[Issue #7556][pulsar-client] Ensure parallel invocations of MultiTopicsConsumerImpl::subscribeAsync with same topic name do not produce an error.

[sandrzejczak]

Fixes #7556

Motivation

We use PatternMultiTopicsConsumerImpl in our solution, but we need to subscribe to new topics instantly after their creation. Default behavior of PatternMultiTopicsConsumerImpl is too slow and too costly for our use case.
This is why we've created an external mechanism which notifies us about new topics. We then subscribe to them using MultiTopicsConsumerImpl#subscribeAsync.
However, if this method is invoked multiple times with same topicName simultaneously, an error may occur and topic's consumers may get closed. It happens, because multiple invocations can pass the initial check in topicNameValid(String topicName), as the topics map does not contain any entry for the topic yet. More detailed description is available at issue's page: #7556

Modifications

Code in MultiTopicsConsumerImpl#doSubscribeTopicPartitions now checks if topics map already contains an entry for a given topic. It does so by checking the return value of putIfAbsent method. If it does already contain an entry, then subscribeResult future is completed exceptionally and the method returns. It prevents subsequent checkState(currentAllTopicsPartitionsNumber == numTopics, "...") invocation from throwing an exception which would cause topic's consumers to get closed.

Verifying this change

Added new unit test to MultiTopicsConsumerImplTest class.

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

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (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? (not applicable)
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation. (not applicable)

[sijie]

/pulsarbot run-failure-checks