Shawyeok opened a new pull request, #26135:
URL: https://github.com/apache/pulsar/pull/26135
Fixes #26125
### Motivation
With a `Shared` subscription, batch-index acknowledgment enabled, and a
`DeadLetterPolicy`
configured, a message that the application explicitly acknowledged on its
final allowed
redelivery round (`redeliveryCount == maxRedeliverCount`) could still be
routed to the DLQ topic.
On every batch redelivery, `ConsumerImpl.receiveIndividualMessagesFromBatch`
creates a fresh
`ackSetInMessageId` bitset (`BatchMessageIdImpl.newAckSet`, all bits
"unacked") that is never
synced against the broker's own `ackSet` for that redelivery. Indices the
broker already knows
are acked are correctly skipped from delivery, but their bit in the fresh
`ackSetInMessageId` is
never cleared. As a result, `MessageIdAdvUtils.acknowledge()` never sees the
batch as fully acked
once some indices were acked in earlier rounds, so
`PersistentAcknowledgmentsGroupingTracker#addIndividualAcknowledgment` never
takes the
"fully acked" branch -- `unAckedMessageTracker.remove()` and
`possibleSendToDeadLetterTopicMessages.remove()` are never called, leaving a
stale DLQ-candidate
list around to be published once the client's own ack-timeout bookkeeping
fires again.
On older client versions where batch-index acknowledgment is not enabled by
default, the same
root cause shows up as a different symptom: the broker never sends a
per-index `ackSet` in this
mode, so `ackSetInMessageId` is the *only* thing that decides when the whole
entry gets acked and
the cursor advances. Because it's rebuilt from scratch on every redelivery,
earlier acks within
the batch are forgotten across rounds and the bitset can permanently fail to
reach "fully acked" --
so even messages that were already published to the DLQ (`ConsumerImpl` acks
the original message
right after a successful DLQ publish) may never actually get marked acked in
the cursor on the
broker side.
### Modifications
- `ConsumerImpl.receiveIndividualMessagesFromBatch`: AND the freshly-created
`ackSetInMessageId`
with the broker-reported `ackSet` bits before processing the batch, so
indices already known to
be acked are correctly reflected from the start of this redelivery round.
- Minor: switch `ackSet` from `List<Long>` to `long[]` end-to-end in
`messageReceived` /
`receiveIndividualMessagesFromBatch` (and the `ZeroQueueConsumerImpl`
override) to avoid
redundant List<->array conversions now that the same array is used for both
`BitSetRecyclable.valueOf` and the new `and(...)` call.
- Added
`DeadLetterTopicTest#testAckedBatchMessageNotSentToDeadLetterTopicOnFinalRedeliveryRound`.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- Added
`DeadLetterTopicTest#testAckedBatchMessageNotSentToDeadLetterTopicOnFinalRedeliveryRound`:
sends a 5-message batch, deliberately lets batch indices 1 and 2 time
out for
`maxRedeliverCount` rounds and acks them on the final allowed round,
then asserts nothing is
ever delivered to the DLQ topic.
- Confirmed this test **fails** without the fix (receives an unexpected
DLQ message) and
**passes** once the fix is applied.
- Manually verified against a live standalone broker with the original
Java reproducer from the
issue, both on a non-partitioned topic and on a 3-partition topic.
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
