fix: introduce ack tracker to prevent loss batch message

[KennyChenFight]

Issue

In the implementation of Pulsar, the producer can enable the batching mechanism to bundle multiple messages into a single batch, sending them all at once to save on RTT. When the consumer receives the batch, it will split the messages within the batch into smaller messages and send them to the client for acknowledgment (ack) or negative acknowledgment (nack).

In the Pulsar Golang client implementation, the ackTracker is used internally to identify which messages belong to the same batch. The consumer waits for all messages within a batch to be acknowledged before sending the acknowledgment to Pulsar.

However, currently, in the Pulsar Golang client implementation, msg.ID().Serialize() does not record the ackTracker. Therefore, when deserializing the message ID, the missing tracker information causes the consumer to be unable to recognize that these messages belong to the same batch. As a result, if a message within a batch is acknowledged first, it will cause other messages within the same batch to also be treated as acknowledged.

This explains why, after a message is nacked, it does not return because the other messages within the same batch have already been acknowledged and will not be redelivered to the consumer.

Solution

Maintain an ackTrackers structure on PulsarConsumerSource to control the batch acknowledgment behavior. Only when all messages within a batch are acknowledged will an acknowledgment be sent to Pulsar.

Note: The current architecture does not support batch index acknowledgment behavior because it would require modifications to the internal behavior of the Pulsar Golang client.

Future Work

Submit a PR to the Pulsar Golang client to make the necessary fixes.

[rueian]

Thank you for looking into pulsar details to find the bug!