[Bug] Dynamically expanding partitions leads to a continuous increase in the number of consumers

[GuanZFF]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

java 1.8
pulsar client 2.7.2
pulsar broker 2.7.2

Minimal reproduce step

The broker notified the consumer to go offline while triggering the execution of the dynamic expansion partition task. Reconnected consumers cannot be restored

What did you expect to see?

No need to exclude consumers, during the process of reconnecting consumers

What did you see instead?

consumers.entrySet().removeIf(e -> TopicName.get(e.getKey()).getPartitionedTopicName().equals(topicName) && !e.getValue().isConnected());

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[GuanZFF]

Reproduction steps:

1. At this point, the breakpoint consumers. introySet(). removeIf (e ->TopicName. get (e.g. etKey()). getPartitionedTopicName(). equals (topicName)&&! e.getValue().isConnected());
2. Execute the reset cursor command
3. MultiTopicscosumerImpl will delete consumers and continuously add consumers corresponding to the last partition