git-enzo opened a new issue #7556:
URL: https://github.com/apache/pulsar/issues/7556
**Describe the bug**
In MultiTopicsConsumerImpl class there is possibility to break an internal
state during subscribe topics. This behavior may occur in
```doSubscribeTopicPartitions()``` method, which is invoking during
subscribing, for example in ```subscribeAsync()``` method.
The MultiTopicsConsumerImpl class stores state about number of partitions
for each subscribed topic. This state is keep in the following map:
```protected final ConcurrentHashMap<String, Integer> topics```. There is also
another variable which keeps information about topic partitions
(```AtomicInteger allTopicPartitionsNumber```), which stores sum of all topic
partitions from all subscribed topics. There is also a map of consumers for
particular topics.
If we are trying to subscribe a new topic, then it is added to topics' map,
and then ```allTopicPartitionsNumber``` counter is incrementing:
```
this.topics.putIfAbsent(topicName, numPartitions);
allTopicPartitionsNumber.addAndGet(numPartitions);
```
or
```
this.topics.putIfAbsent(topicName, 1);
allTopicPartitionsNumber.incrementAndGet();
```
The new consumer will be also added to consumer's map.
The problem may occur, when two _same topics_ (topic names) will be
subscribed at _the same time_. In this case in one of the subscribe operation,
the topic will be not added to topics' map, because of ```putIfAbsent()```
method, but counter will be always incremented. If we assume that two separate
subscription operations are running at the same time, for example due to
asynchronous, then both of them may failed, because state will be incorrect:
```
int numTopics =
this.topics.values().stream().mapToInt(Integer::intValue).sum();
checkState(allTopicPartitionsNumber.get() == numTopics,
"allTopicPartitionsNumber " +
allTopicPartitionsNumber.get()
+ " not equals expected: " + numTopics);
```
This behavior cause closing internal consumer, which will be also removed
from consumer's map and topic will be not subscribed.
I think this is an unexpected behavior, because if we run such two
operations sequentially, then everything would be fine - topic will be
subscribed. To sum up, as ```doSubscribeTopicPartitions()``` method is invoked
from ```subscribeAsync()``` method, which should be asynchronous, then such
behavior is unexpected.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
