[Bug] Ack failure on message listener in multi topics consumer

[nkurihar]

Search before asking

• [X] I searched in the issues and found nothing similar.

Version

main branch commit id: 8b2753a56579ea6cf11e26a0c5a160797518df63

Minimal reproduce step

1. publish a certain amount of messages to the target topic
2. start a "multi topic" (e.g. partitioned, list, regex) consumer with message listener
3. ack failure with AlreadyClosed occurs

• The likelihood of occurrence may be increased by restarting the consumer several times with continuously publishing messages.
• The smallest code for consumer: https://gist.github.com/nkurihar/bf0bd8a56cf0ba30530488d9db0e63bd (https://github.com/apache/pulsar-client-cpp/blob/main/examples/SampleConsumerListener.cc plus outputs error log on ack failure)

example:

// publishing messages continuously
$ pulsar-perf produce -s 1 persistent://my-tenant/my-ns/my-partitioned-topic

// restart consumer several times
$ ./SampleConsumerListener 
^C
$ ./SampleConsumerListener
^C
.
.
.
2024-10-02 17:48:13.309 INFO  [0x16afff000] MultiTopicsConsumerImpl:308 | Successfully Subscribed to a single partition of topic in TopicsConsumer. Partitions need to create : 3
2024-10-02 17:48:13.309 INFO  [0x16afff000] HandlerBase:138 | Finished connecting to broker after 218 ms
2024-10-02 17:48:13.309 ERROR [0x16b117000] SampleConsumerListener:34 | Failed to ack: AlreadyClosed, Message(prod=xxx, seq=70, publish_time=1727858342279, payload_size=1, msg_id=(62407117,85,0,0), props={})
2024-10-02 17:48:13.309 ERROR [0x16b117000] SampleConsumerListener:34 | Failed to ack: AlreadyClosed, Message(prod=xxx, seq=71, publish_time=1727858342439, payload_size=1, msg_id=(62407117,86,0,0), props={})
2024-10-02 17:48:13.309 ERROR [0x16b117000] SampleConsumerListener:34 | Failed to ack: AlreadyClosed, Message(prod=xxx, seq=72, publish_time=1727858342599, payload_size=1, msg_id=(62407117,87,0,0), props={})

What did you expect to see?

All messages should be acked successfully

What did you see instead?

Failed to ack some messages

Anything else?

No response

Are you willing to submit a PR?

• [X] I'm willing to submit a PR!