Skip to content

[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages#604

Open
geniusjoe wants to merge 1 commit into
apache:mainfrom
geniusjoe:dev/chunk-nack-acktimeout
Open

[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages#604
geniusjoe wants to merge 1 commit into
apache:mainfrom
geniusjoe:dev/chunk-nack-acktimeout

Conversation

@geniusjoe

Copy link
Copy Markdown
Contributor

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

  1. 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.

  2. remove() method — Apply the same logic as add() to ensure consistent key lookup and removal from the tracker.

tests/MessageChunkingTest.cc

  1. testNegativeAckChunkedMessage — Verifies that a chunked message can be correctly redelivered after nack (aligned with Go TestChunkAckAndNAck and Java testNegativeAckChunkedMessage).

  2. testAckTimeoutChunkedMessage — Verifies that a chunked message can be correctly redelivered after ack timeout (aligned with Java testLargeMessageAckTimeOut).

  3. 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

geniusjoe#2
image

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant