[bug][cpp] Fix issue where unexpected ack timeout occurred

[massakam]

Motivation

When we start a consumer like the following and publish multiple messages to a topic, ack timeout occurs even though the messages received are immediately acknowledged. This does not occur with normal topics, only with partitioned topics.

ClientConfiguration clientConfig;
clientConfig.setMessageListenerThreads(1);
Client client("pulsar://localhost:6650", clientConfig);

ConsumerConfiguration consumerConfig;
consumerConfig.setConsumerType(ConsumerShared);
consumerConfig.setUnAckedMessagesTimeoutMs(10000);
consumerConfig.setMessageListener([](Consumer con, const Message& msg) {
    cout << "Received: " << msg.getDataAsString() << endl;
    con.acknowledge(msg);
    this_thread::sleep_for(chrono::milliseconds(20000));
});

Consumer consumer;
client.subscribe("persistent://public/default/pt4", "sub1", consumerConfig, consumer);

The reason is that the task of executing the message listener function is queued to listenerExecutor_ immediately after adding the message to unAckedMessageTrackerPtr_, but it is not always executed immediately. If it takes a long time to execute each message listener function like the code for reproduction above, it will take a long time for the queued tasks to actually execute, causing an ack timeout. I don't think this is the expected behavior.

pulsar/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc

Lines 462 to 464 in 0bbc4e1

	unAckedMessageTrackerPtr_->add(msg.getMessageId());
	listenerExecutor_->postWork(
	std::bind(&MultiTopicsConsumerImpl::internalListener, shared_from_this(), consumer));

Modifications

Have listenerExecutor_ perform adding messages to unAckedMessageTrackerPtr_. In this way, the message listener function will always be executed immediately after adding a message to unAckedMessageTrackerPtr_, preventing unexpected ack timeouts.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc-not-needed

[BewareMyPower]

LGTM. Nice catch!

[github-actions]

@massakam Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

[shibd]

Nice catch!