[fix][consumer] Initialize batch ack tracker from the broker ack set on redelivery

[Shawyeok]

Motivation

This ports the fix from apache/pulsar#26135 ([fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ) to the Go client.

When the broker redelivers a batch entry that was partially acked with batch index acknowledgment, it attaches the ack set of the already-acked indexes to the CommandMessage. The Go client uses that ack set to skip delivering the acked indexes, but initializes the batch's local ackTracker as if every index were still outstanding.

For a consumer without batch index ack this is fatal: such a consumer can only ack a batch as a whole entry, once the tracker reports every index acked. A stale tracker can never complete — the broker-acked indexes are never delivered again, so their bits are never cleared — and ackIDCommon silently drops the acks of the redelivered messages. The batch is never acked at the broker and its messages are redelivered to the next consumer session; with a dead letter policy their redeliveries eventually exhaust, and messages the application acked in time end up in the DLQ.

(For a consumer with batch index ack the stale tracker is harmless: each ack is sent as an individual batch-index ack and the broker intersects ack sets, so the broker state converges regardless. This is why the regression test below runs the final-round acks through a consumer without batch index ack.)

Modifications

• newAckTracker takes the set of outstanding batch indexes as a parameter, and the consumer seeds it with the broker-provided ack set of the redelivered batch when one is present (nil keeps the previous all-outstanding behavior).
• Add TestAckTrackerWithBatchIDs, dedicated unit tests for the amended newAckTracker: seeding from a broker ack set, completion by acking only the outstanding indexes (individually and cumulatively), clone semantics of the passed bitset, and the nil contract.
• Add TestAckedBatchMessageNotSentToDeadLetterTopicWithoutBatchIndexAck, a Go adaptation of testAckedBatchMessageNotSentToDeadLetterTopicOnFinalRedeliveryRound from the original PR: batch indexes 1 and 2 of a partially-acked batch are nacked until their final allowed redelivery round, acked there, and the test asserts that nothing is routed to the DLQ. Two Go-specific adaptations: redelivery is driven by Nack (the Go client has no ackTimeout), and the final application-visible round is redeliveryCount == MaxDeliveries-1, since the Go client routes messages that reach MaxDeliveries straight to the DLQ router without delivering them to the application.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Against a local Pulsar standalone (apachepulsar/pulsar:4.0.10 with integration-tests/conf/standalone.conf):

  go test ./pulsar -run '^TestAckedBatchMessageNotSentToDeadLetterTopicWithoutBatchIndexAck$' -count=1
  

  passes with the fix. With the fix reverted (pulsar/consumer_partition.go from master), the test fails — messages 1 and 2, acked on their final redelivery round, are routed to the DLQ:

  Messages: no message should have been routed to the DLQ, but received: ... ORIGIN_MESSAGE_ID:11:0:0:1 ...
  

• go test ./pulsar -run '^TestBatchIndexAck$' -count=1 still passes.

• Unit tests: go test ./pulsar -run '^TestAckTracker$|^TestAckTrackerWithBatchIDs$|^TestAckingMessageIDBatchOne$|^TestAckingMessageIDBatchTwo$' -count=1 (no broker required).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API: no
• The schema: no
• The default values of configurations: no
• The wire protocol: no

Documentation

• Does this pull request introduce a new feature? no

🤖 Generated with Claude Code