regex subscription not working for new topics in Python

[baynes]

Describe the bug
If a new topic is added that matches a regex subscription then if the client is written in Java it detects it and adds a cursor but if it is written in Python it does not.

To Reproduce
Steps to reproduce the behavior:

Run the following as a Phython3 client:

#!/usr/bin/python3
import pulsar
import re

client = pulsar.Client("pulsar://localhost:6650")
initial_position=pulsar.InitialPosition.Earliest )
consumer = client.subscribe(re.compile('.*'), subscription_name='my-sub' )

while True:
    msg = consumer.receive()
    print("Received message '%s'" % msg.data())
    consumer.acknowledge(msg)

client.close()

In another window run this command to get a Java client for comparison:

/opt/pulsar/bin/pulsar-client consume --regex '.*' -s all -n 0

In another window send a message to a new topic:

/opt/pulsar//bin/pulsar-client produce addtopic -m 'm1'

Wait a minute for the clients to detect the new topic.
Send another message to the same topic.

/opt/pulsar//bin/pulsar-client produce addtopic -m 'm2'

The Java client receives the message but the Python one does not.

Kill and restart the clients. Send a message to the same topic.

/opt/pulsar//bin/pulsar-client produce addtopic -m 'm3'

Both clients receive the message.

Expected behavior
Both clients should receive the second message.

Screenshots
NA

Desktop (please complete the following information):
Centos 7
Pulsar 2.5.1
Python pulsar-client 2.5.2

Additional context
initial_position=pulsar.InitialPosition.Earliest does not help

[jiazhai]

Thanks @baynes for reporting this issue. Since it is able to reproduce, mark it as help-wanted firstly. Any help on fix this issue is very appreciated.

[BewareMyPower]

TL;DR It's because there's a deadlock in C++ client. I'll push a PR soon.

Just because C++ stand library doesn't have something like ConcurrentHashMap, the current implementation used a mutex_ to share a std::map.
However, the mutex_ is also acquired in receive, so if you called receive, the mutex_ is acquired and held until a new message was pushed to the internal message queue. Then if the topic discovery timer found new topics, the callback to add new topics also needed to acquire the mutex_, which leads to a deadlock.

The deadlock problem is similar to my PR before, see the change of PartitionedConsumerImpl::receive in #6732.