You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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
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.
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.
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
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.
remove() method — Apply the same logic as add() to ensure consistent key lookup and removal from the tracker.
tests/MessageChunkingTest.cc
testNegativeAckChunkedMessage — Verifies that a chunked message can be correctly redelivered after nack (aligned with Go TestChunkAckAndNAck and Java testNegativeAckChunkedMessage).
testAckTimeoutChunkedMessage — Verifies that a chunked message can be correctly redelivered after ack timeout (aligned with Java testLargeMessageAckTimeOut).
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.
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!
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!
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 file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Prior to this change, the C++ client did not properly handle chunked messages in the following scenarios:
ChunkMessageIdwas 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.UnAckedMessageTrackercalleddiscardBatch()onChunkMessageIdImpl, which created a new plainMessageIdImpland lost all internal chunk entry information. When ack timeout triggered, the tracker could not expand chunk entries for redelivery.This aligns the C++ client behavior with the Go and Java clients, which already handle these scenarios correctly.
Modifications
lib/ConsumerImpl.ccnegativeAcknowledge()— When the MessageId is aChunkMessageIdImpl, expand all chunk entries viagetChunkedMessageIds()and add each one individually tonegativeAcksTracker_. This ensures the broker redelivers all chunks.redeliverMessages()— Before sendingRedeliverUnacknowledgedMessagesto the broker, expand anyChunkMessageIdImplin the input set into all its chunk entries. This is the path triggered by ack timeout.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.ccadd()method — SkipdiscardBatch()forChunkMessageIdImpl.discardBatch()creates a new plainMessageIdImpl, which would lose thechunkedMessageIds_vector stored insideChunkMessageIdImpl. By preserving the original object, when ack timeout triggers and the MessageId is passed toredeliverMessages(), it can be dynamically cast back toChunkMessageIdImpland expanded into all chunk entries.remove()method — Apply the same logic asadd()to ensure consistent key lookup and removal from the tracker.tests/MessageChunkingTest.cctestNegativeAckChunkedMessage— Verifies that a chunked message can be correctly redelivered after nack (aligned with GoTestChunkAckAndNAckand JavatestNegativeAckChunkedMessage).testAckTimeoutChunkedMessage— Verifies that a chunked message can be correctly redelivered after ack timeout (aligned with JavatestLargeMessageAckTimeOut).testChunkedMessageDLQ— Verifies that a chunked message is correctly sent to the DLQ topic after exceedingmaxRedeliverCount, with complete message content and correctPROPERTY_ORIGIN_MESSAGE_IDandSYSTEM_PROPERTY_REAL_TOPICproperties.Design Details
Why skip
discardBatch()forChunkMessageIdImpl?discardBatch()creates a newMessageIdImplretaining onlyledgerId,entryId, andpartitionIndex, discardingbatchIndexandbatchSize. ForChunkMessageIdImpl(a subclass ofMessageIdImpl), this would also lose thechunkedMessageIds_vector that stores all chunk entry positions. Without this information, the client cannot expand chunk entries for redelivery when ack timeout fires.Note:
ChunkMessageIdImplalready hasbatchIndex = -1(chunked messages are not batched), so skippingdiscardBatch()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
ChunkMessageIdImpl.chunkedMessageIds_unAckedChunkedMessageIdSequenceMapunAckChunksTrackernegativeAcknowledge()entryNegativeAcksTrackertimeoutNackID()entryredeliverMessages()UnAckedMessageTrackertimeout callbackdynamic_pointer_castcheckinstanceof ChunkMessageIdImplchecksetChunkingEnabled(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
geniusjoe#2
