You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "geniusjoe (via GitHub)" <gi...@apache.org> on 2026/07/17 16:53:08 UTC
[PR] [fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages [pulsar-client-cpp]
geniusjoe opened a new pull request, #604:
URL: https://github.com/apache/pulsar-client-cpp/pull/604
## Motivation
Prior to this change, the C++ client did not properly handle chunked messages in the following scenarios:
- **Negative acknowledgement (nack)**: Only the assembled `ChunkMessageId` was nacked, without expanding all chunk entries. The broker would only redeliver the last chunk entry, making it impossible for the consumer to reassemble the complete message.
- **Ack timeout redelivery**: `UnAckedMessageTracker` called `discardBatch()` on `ChunkMessageIdImpl`, which created a new plain `MessageIdImpl` and lost all internal chunk entry information. When ack timeout triggered, the tracker could not expand chunk entries for redelivery.
- **Dead Letter Queue (DLQ)**: The DLQ producer did not enable chunking, so large messages could not be correctly sent to the DLQ topic.
This aligns the C++ client behavior with the Go and Java clients, which already handle these scenarios correctly.
## Modifications
### `lib/ConsumerImpl.cc`
1. **`negativeAcknowledge()`** — When the MessageId is a `ChunkMessageIdImpl`, expand all chunk entries via `getChunkedMessageIds()` and add each one individually to `negativeAcksTracker_`. This ensures the broker redelivers all chunks.
2. **`redeliverMessages()`** — Before sending `RedeliverUnacknowledgedMessages` to the broker, expand any `ChunkMessageIdImpl` in the input set into all its chunk entries. This is the path triggered by ack timeout.
3. **DLQ producer configuration** — Enable `setChunkingEnabled(true)` for the DLQ producer so that large messages can be correctly chunked when sent to the DLQ topic.
### `lib/UnAckedMessageTrackerEnabled.cc`
4. **`add()` method** — Skip `discardBatch()` for `ChunkMessageIdImpl`. `discardBatch()` creates a new plain `MessageIdImpl`, which would lose the `chunkedMessageIds_` vector stored inside `ChunkMessageIdImpl`. By preserving the original object, when ack timeout triggers and the MessageId is passed to `redeliverMessages()`, it can be dynamically cast back to `ChunkMessageIdImpl` and expanded into all chunk entries.
5. **`remove()` method** — Apply the same logic as `add()` to ensure consistent key lookup and removal from the tracker.
### `tests/MessageChunkingTest.cc`
6. **`testNegativeAckChunkedMessage`** — Verifies that a chunked message can be correctly redelivered after nack (aligned with Go `TestChunkAckAndNAck` and Java `testNegativeAckChunkedMessage`).
7. **`testAckTimeoutChunkedMessage`** — Verifies that a chunked message can be correctly redelivered after ack timeout (aligned with Java `testLargeMessageAckTimeOut`).
8. **`testChunkedMessageDLQ`** — Verifies that a chunked message is correctly sent to the DLQ topic after exceeding `maxRedeliverCount`, with complete message content and correct `PROPERTY_ORIGIN_MESSAGE_ID` and `SYSTEM_PROPERTY_REAL_TOPIC` properties.
## Design Details
### Why skip `discardBatch()` for `ChunkMessageIdImpl`?
`discardBatch()` creates a new `MessageIdImpl` retaining only `ledgerId`, `entryId`, and `partitionIndex`, discarding `batchIndex` and `batchSize`. For `ChunkMessageIdImpl` (a subclass of `MessageIdImpl`), this would also lose the `chunkedMessageIds_` vector that stores all chunk entry positions. Without this information, the client cannot expand chunk entries for redelivery when ack timeout fires.
Note: `ChunkMessageIdImpl` already has `batchIndex = -1` (chunked messages are not batched), so skipping `discardBatch()` does not affect the key normalization purpose (deduplicating batch sub-entries in the tracker).
### Broker perspective
The broker does not recognize chunk message semantics. Each chunk is an independent entry in the managed ledger. The client is responsible for expanding all chunk entries when requesting redelivery, so the broker can replay all positions belonging to the same chunked message.
### Comparison with Go and Java clients
| Aspect | C++ (this PR) | Java | Go |
|--------|---------------|------|----|
| Chunk info storage | Inside `ChunkMessageIdImpl.chunkedMessageIds_` | External map `unAckedChunkedMessageIdSequenceMap` | External `unAckChunksTracker` |
| Nack expand timing | At `negativeAcknowledge()` entry | At `NegativeAcksTracker` timeout | At `NackID()` entry |
| AckTimeout expand timing | In `redeliverMessages()` | In `UnAckedMessageTracker` timeout callback | N/A (Go relies on nack) |
| Skip discardBatch for chunk | `dynamic_pointer_cast` check | `instanceof ChunkMessageIdImpl` check | Type assertion at call site |
| DLQ producer chunking | `setChunkingEnabled(true)` | `chunkingEnabled(true)` | `WithChunking(true)` |
All three clients achieve the same result: expand all chunk entries and send them to the broker for complete redelivery.
## Local workflow
https://github.com/geniusjoe/pulsar-client-cpp/pull/2
<img width="2638" height="945" alt="image" src="https://github.com/user-attachments/assets/9dfa5f0b-dbf0-41b3-ac60-bb26cd30ec4f" />
--
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] Support nack, ackTimeout redelivery and dlq for chunked messages [pulsar-client-cpp]
Posted by "oversearch (via GitHub)" <gi...@apache.org>.
oversearch commented on PR #604:
URL: https://github.com/apache/pulsar-client-cpp/pull/604#issuecomment-5028285160
Just a heads up - I was testing out chunked messages from the C++ API today, and noticed that the `ReaderConfiguration` class is missing all of the chunking related options that the Consumer class has. It's not directly related to this change, but is another gap in the chunked message functionality... not sure if you'd like to cover that here. If not, I might get some time to implement it in the next few weeks, as my company is planning to start using this feature. Thanks!
--
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] Support nack, ackTimeout redelivery and dlq for chunked messages [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on PR #604:
URL: https://github.com/apache/pulsar-client-cpp/pull/604#issuecomment-5030437599
> Just a heads up - I was testing out chunked messages from the C++ API today, and noticed that the `ReaderConfiguration` class is missing all of the chunking related options that the Consumer class has. It's not directly related to this change, but is another gap in the chunked message functionality... not sure if you'd like to cover that here. If not, I might get some time to implement it in the next few weeks, as my company is planning to start using this feature. Thanks!
@oversearch
Thanks for pointing this out!
I don't have immediate plans to add chunk configuration support to `ReaderConfiguration`, as my primary use case is based on shared subscription mode and I'm not very familiar with the cumulative ack logic that Reader relies on.
However, I agree it makes sense to add these options — since Reader is internally backed by an exclusive consumer, the chunk-related parameters (`MaxPendingChunkedMessage`, `ExpireTimeOfIncompleteChunkedMessageMs`, `AutoAckOldestChunkedMessageOnQueueFull`) should be configurable for Reader as well.
--
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] Support nack, ackTimeout redelivery and dlq for chunked messages [pulsar-client-cpp]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #604:
URL: https://github.com/apache/pulsar-client-cpp/pull/604
--
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