[C++] PartitionedConsumerImpl send redeliverMessages repeatedly

[saosir]

Describe the bug
Both PartitionedConsumerImpl and ConsumerImpl have member variable unAckedMessageTrackerPtr_ (class UnAckedMessageTrackerEnabled), and PartitionedConsumerImpl is composed of ConsumerImpl. If the acknowledgement times out, they will send redeliverMessages repeatedly, MultiTopicsConsumerImpl has same problem.

see #8606 (comment)

To Reproduce

1. subscribe a partitioned topic and set UnAckedMessagesTimeoutMs

Consumer consumer;
ConsumerConfiguration consumerConf;
consumerConf.setUnAckedMessagesTimeoutMs(11000);  // must >= 10000
consumerConf.setTickDurationInMs(11000);
Result result = client.subscribe("my-topic", "consumer-1", consumerConf, consumer);

2. recevie a msg from partitioned topic and not do acknowledge

Message msg;
consumer.recevie(msg)
sleep(20000);

3. sleep 20000ms to trigger redeliverMessages repeatedly

sleep(20000);

Expected behavior
Only one redeliverUnacknowledgedMessages should be sent in PartitionedConsumerImpl

[saosir]

Is the member variable unAckedMessageTrackerPtr_ in PartitionedConsumerImpl and MultiTopicsConsumerImpl redundant?

[codelipenghui]

@BewareMyPower Please help take a look at this issue.

[BewareMyPower]

@codelipenghui I agree with @saosir , the point is that should Java client's implementation also have the problem? See

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

Line 101 in c17253b

private final UnAckedMessageTracker unAckedMessageTracker;

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 134 in c17253b

private final UnAckedMessageTracker unAckedMessageTracker;

Both ConsumerImpl and MultiTopicsConsumerImpl hold an UnAckedMessageTracker, when the tracker's timeout exceeds, it will trigger the redelivery.

IMO, the unAckedMessageTracker of MultiTopicsConsumerImpl is redundant. And this feature of C++ client is ported from Java client. So I think if C++ client needs changes, Java client also needs changes.

[saosir]

@BewareMyPower ConsumerImpl will check if there is a parent when track message in Java client's implementation

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Lines 1541 to 1547 in c17253b

	if (hasParentConsumer) {
	//TODO: check parent consumer here
	// we should no longer track this message, TopicsConsumer will take care from now onwards
	unAckedMessageTracker.remove(id);
	} else {
	unAckedMessageTracker.add(id);
	}

[saosir]

In addition, UnAckedMessageTrackerEnable did not clean up the batchIndex of MessageId when tracking the message. That will cause only the last message in the batch to be cleared from UnAckedMessageTrackerEnabled when doAcknowledgeIndividual and send redeliverMessages repeatedly

[BewareMyPower]

Got it. It looks like we should also check if there's a parent in C++ client.

[BewareMyPower]

Since the reason is clear, could you contribute a PR to fix it?

[saosir]

OK, I will submit a PR later