You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Shawyeok (via GitHub)" <gi...@apache.org> on 2026/07/02 04:16:11 UTC
[PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4875910766
Since other language clients (pulsar-client-cpp, pulsar-client-go, pulsar-dotnet) seem to follow Pulsar Java client's implementation details, a similar bug might exist there. @Shawyeok Would you like to check pulsar-client-cpp and pulsar-client-go?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26135:
URL: https://github.com/apache/pulsar/pull/26135
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4876061925
> Since other language clients (pulsar-client-cpp, pulsar-client-go, pulsar-dotnet) seem to follow Pulsar Java client's implementation details, a similar bug might exist there. @Shawyeok Would you like to check pulsar-client-cpp and pulsar-client-go?
Surely, I'll take a look.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4873798919
> > The main fix in `receiveIndividualMessagesFromBatch` looks correct to me: syncing the broker-provided ack set into `ackSetInMessageId` fixes the normal batch path.
> > However, I think the coverage/fix is incomplete. When a consumer is configured with `messagePayloadProcessor(...)`, `ConsumerImpl.messageReceived` returns through `processPayloadByProcessor(...)`, and `MessagePayloadContextImpl.get(...)` still creates a fresh all-unacked `ackSetInMessageId`. It only uses the broker ack set for `ackBitSet`, so previously acked/skipped batch indexes are not reflected in the `BatchMessageIdImpl`.
> > That means consumers using `messagePayloadProcessor(MessagePayloadProcessor.DEFAULT)` or a custom processor can still hit the same fully-acked-batch detection failure, leaving `unAckedMessageTracker` / `possibleSendToDeadLetterTopicMessages` uncleared.
> > Could we apply the same `ackSetInMessageId.and(BitSet.valueOf(ackSet))` logic to the payload-processor path as well, preferably via a shared helper, and add a regression variant that sets `.messagePayloadProcessor(MessagePayloadProcessor.DEFAULT)`?
>
> Sounds reasonable. I'll fix it soon.
@void-ptr974 Fixed, feel free to take a look.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on code in PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#discussion_r3517844240
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1796,8 +1779,9 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
BitSet ackSetInMessageId = BatchMessageIdImpl.newAckSet(batchSize);
BitSetRecyclable ackBitSet = null;
- if (ackSet != null && ackSet.size() > 0) {
- ackBitSet = BitSetRecyclable.valueOf(SafeCollectionUtils.longListToArray(ackSet));
+ if (ackSet != null && ackSet.length > 0) {
+ ackBitSet = BitSetRecyclable.valueOf(ackSet);
+ ackSetInMessageId.and(BitSet.valueOf(ackSet));
Review Comment:
Nope, `ackBitSet` is an instance of `BitSetRecyclable`, the class `BitSetRecyclable` isn't a subclass of `BitSet`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on code in PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#discussion_r3519695923
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1796,8 +1779,9 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
BitSet ackSetInMessageId = BatchMessageIdImpl.newAckSet(batchSize);
BitSetRecyclable ackBitSet = null;
- if (ackSet != null && ackSet.size() > 0) {
- ackBitSet = BitSetRecyclable.valueOf(SafeCollectionUtils.longListToArray(ackSet));
+ if (ackSet != null && ackSet.length > 0) {
+ ackBitSet = BitSetRecyclable.valueOf(ackSet);
+ ackSetInMessageId.and(BitSet.valueOf(ackSet));
Review Comment:
That's true, thanks for point it out, fixed.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4873153126
> The main fix in `receiveIndividualMessagesFromBatch` looks correct to me: syncing the broker-provided ack set into `ackSetInMessageId` fixes the normal batch path.
>
> However, I think the coverage/fix is incomplete. When a consumer is configured with `messagePayloadProcessor(...)`, `ConsumerImpl.messageReceived` returns through `processPayloadByProcessor(...)`, and `MessagePayloadContextImpl.get(...)` still creates a fresh all-unacked `ackSetInMessageId`. It only uses the broker ack set for `ackBitSet`, so previously acked/skipped batch indexes are not reflected in the `BatchMessageIdImpl`.
>
> That means consumers using `messagePayloadProcessor(MessagePayloadProcessor.DEFAULT)` or a custom processor can still hit the same fully-acked-batch detection failure, leaving `unAckedMessageTracker` / `possibleSendToDeadLetterTopicMessages` uncleared.
>
> Could we apply the same `ackSetInMessageId.and(BitSet.valueOf(ackSet))` logic to the payload-processor path as well, preferably via a shared helper, and add a regression variant that sets `.messagePayloadProcessor(MessagePayloadProcessor.DEFAULT)`?
Sounds reasonable. I'll fix it soon.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#discussion_r3519698434
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java:
##########
@@ -341,4 +351,49 @@ public void testUpdateAutoScaleReceiverQueueHintRaceWithConcurrentDrain() {
"Hint must reflect the post-enqueue state (pipeline had >=1 message); "
+ "a concurrent drain of the just-enqueued message must not clear it.");
}
+
+ @Test(invocationTimeOut = 1000)
+ public void testGetMessageAtSyncsAckSetInMessageIdWithBrokerAckSet() {
+ // Regression test for MessagePayloadContextImpl#getMessageAt: the BatchMessageIdImpl handed
+ // back to the caller carries a shared ackSetInMessageId bitset that must be seeded from the
+ // broker-reported ackSet, not a fresh "all unacked" bitset. Otherwise indices the broker
+ // already knows are acked would be reported as still-outstanding in the returned MessageId,
+ // which is the same root cause that let acked batch messages leak into the DLQ (see
+ // ConsumerImpl#receiveIndividualMessagesFromBatch and its ackSetInMessageId.and(...) fix).
+ final int batchSize = 3;
+ MessageMetadata messageMetadata = new MessageMetadata()
+ .setProducerName("test-producer")
+ .setSequenceId(0)
+ .setPublishTime(System.currentTimeMillis())
+ .setNumMessagesInBatch(batchSize);
+
+ // Broker reports index 0 as already acked (bit cleared); indices 1 and 2 are still
+ // outstanding (bits set). This mirrors the ackSet the broker attaches on redelivery.
+ BitSet brokerAckSet = new BitSet(batchSize);
+ brokerAckSet.set(1);
+ brokerAckSet.set(2);
+ List<Long> ackSet = Arrays.stream(brokerAckSet.toLongArray()).boxed().collect(Collectors.toList());
+
+ MessageIdImpl messageId = new MessageIdImpl(1L, 2L, -1);
+ MessagePayloadContextImpl context = MessagePayloadContextImpl.get(
+ null, messageMetadata, messageId, consumer, 0, ackSet, DEFAULT_CONSUMER_EPOCH);
+ try {
+ // Index 0 is already acked per the broker, so it must not be redelivered to the app.
+ MessagePayload payload0 = MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{0}));
+ Assert.assertNull(context.getMessageAt(0, batchSize, payload0, false, Schema.BYTES));
+
+ MessagePayload payload1 = MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{1}));
+ Message<byte[]> message1 = context.getMessageAt(1, batchSize, payload1, false, Schema.BYTES);
+ Assert.assertNotNull(message1);
Review Comment:
Could use `@Cleanup("release")` for both `MessagePayload` instances to avoid the leak that @void-ptr974 pointed out.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4875805752
The functional fix looks good now. One issue in the new test: `payload0` and `payload1` are created with `MessagePayloadImpl.create(...)` but are never released. `getMessageAt(...)` only releases the retained buffer it creates internally, not the original `MessagePayloadImpl`, so this test leaks ref-counted buffers.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on code in PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#discussion_r3519695923
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1796,8 +1779,9 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
BitSet ackSetInMessageId = BatchMessageIdImpl.newAckSet(batchSize);
BitSetRecyclable ackBitSet = null;
- if (ackSet != null && ackSet.size() > 0) {
- ackBitSet = BitSetRecyclable.valueOf(SafeCollectionUtils.longListToArray(ackSet));
+ if (ackSet != null && ackSet.length > 0) {
+ ackBitSet = BitSetRecyclable.valueOf(ackSet);
+ ackSetInMessageId.and(BitSet.valueOf(ackSet));
Review Comment:
That's true, thanks for point it out, fixed.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ [pulsar]
Posted by "Shawyeok (via GitHub)" <gi...@apache.org>.
Shawyeok commented on PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#issuecomment-4875878010
That's true, thanks for point it out, fixed.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org