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/06/01 14:50:02 UTC
[PR] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
geniusjoe opened a new pull request, #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587
Master Issue: https://github.com/apache/pulsar/pull/13627
Related Issue: apache/pulsar#21070 and apache/pulsar#21101
### Motivation
apache/pulsar#21070 and apache/pulsar#21101 fixed two critical issues in the Java client's chunked message
handling:
1. **Unable to reassemble chunked messages after redeliver**: When a chunked message is redelivered
(e.g., due to broker unload or reconnect), the consumer receives duplicated chunks. The old
code could not handle this correctly:
- For duplicated first chunk (chunkId=0): the old context was not properly cleaned up and restarted,
causing the message to never be assembled.
- For duplicated middle chunks: the chunk would be rejected (since chunkId ≤ lastChunkedMessageId),
and the old code would discard the context entirely, making the message unrecoverable.
2. **Ack holes caused by corrupted or orphaned chunks**: When a different producer reuses the same uuid
(corrupted chunk scenario), or when chunk context is discarded due to gap/expiration, the stale cached
chunks or the incoming corrupted chunks were never acknowledged. This causes the broker subscription
cursor to get stuck, leading to message backlog accumulation that never drains — even after all
logically valid messages have been consumed and acknowledged.
These PRs added logic to distinguish between redeliver (same messageId) and corruption (different
messageId), allowing the consumer to correctly restart chunk assembly on redeliver while acking stale
chunks on corruption to prevent ack holes.
The C++ client had the same issues. This PR ports the equivalent logic to ensure consistent behavior
across all client implementations.
**Note**: Currently, after a chunked message is assembled, the ackTimeout and nack logic only tracks/handles
the last chunk message (i.e., the final messageId of the assembled message). This means if ackTimeout or
nack triggers a redeliver, only the last chunk entry is redelivered rather than all chunk entries. This
limitation needs to be addressed in a follow-up PR.
### Modifications
**Core logic changes in `ConsumerImpl.cc` (`processMessageChunk`)**:
- **Part 1 (chunkId == 0)**: When receiving a duplicated first chunk for a uuid that already has an
incomplete context, detect whether it's a redeliver (same messageId in cache) or corruption (different
messageId). For redeliver: remove old context and restart assembling. For corruption: ack all cached
chunks to avoid ack holes, then restart.
- **Part 3 (duplicated middle chunk)**: When receiving a chunk with chunkId ≤ lastChunkedMessageId,
detect whether it's a redeliver or corruption. For redeliver: simply discard the duplicate and continue
waiting for the next expected chunk. For corruption: ack the corrupted chunk to avoid ack holes.
- **Part 3 (gap chunk)**: When receiving a chunk that skips expected sequence numbers, ack the chunk if
it has expired to avoid ack holes.
- Added `LOG_WARN` and `LOG_INFO` for observability across all scenarios.
- Added detailed comments explaining each part of the chunk processing logic with examples.
**Test changes in `MessageChunkingTest.cc`**:
- Added `testResendChunkMessagesWithoutAckHole`: Verifies that resending the first chunk (chunkId=0)
allows correct reassembly without ack holes.
- Added `testResendChunkMessages`: Verifies interleaved chunk resends across multiple uuids assemble
correctly.
- Added `testResendChunkWithAckHoleMessages`: Verifies duplicated middle chunks are filtered correctly
and chunk gaps cause context cleanup.
- Refactored existing tests to reuse the `sendSingleChunk` helper function for better readability.
### Verifying this change
This change added tests and can be verified as follows:
- Added `testResendChunkMessagesWithoutAckHole` to verify correct reassembly when first chunk is resent,
checking broker backlog drops to 0 after ack.
- Added `testResendChunkMessages` to verify correct assembly with interleaved resends across multiple uuids.
- Added `testResendChunkWithAckHoleMessages` to verify duplicated middle chunks are filtered and gap
chunks cause context cleanup.
### Documentation
- [x] `doc-not-needed`
(This is a bug fix for chunk message handling internals with no user-facing API changes)
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie merged PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3339358227
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
Review Comment:
Fixed. Added increaseAvailablePermits(cnx) for the last chunk case to prevent permit leak.
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie closed pull request #587: [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks
URL: https://github.com/apache/pulsar-client-cpp/pull/587
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3339610936
##########
tests/MessageChunkingTest.cc:
##########
@@ -96,6 +96,18 @@ class MessageChunkingTest : public ::testing::TestWithParam<CompressionType> {
std::string MessageChunkingTest::largeMessage = createLargeMessage();
+// Helper function: send a single chunk message
+static void sendSingleChunk(Producer& producer, const std::string& uuid, int chunkId, int totalChunks) {
+ std::string content = "chunk-" + uuid + "-" + std::to_string(chunkId) + "|";
+ auto msg = MessageBuilder().setContent(content).build();
+ auto& metadata = PulsarFriend::getMessageMetadata(msg);
+ metadata.set_num_chunks_from_msg(totalChunks);
+ metadata.set_chunk_id(chunkId);
+ metadata.set_uuid(uuid);
+ metadata.set_total_chunk_msg_size(100);
+ producer.send(msg);
Review Comment:
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3559007275
##########
lib/ConsumerImpl.cc:
##########
@@ -512,47 +563,117 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
Review Comment:
Fixed by adding trackMessage(messageId) for unexpired chunks, aligned with the referenced Java client logic.
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3339535482
##########
lib/ConsumerImpl.cc:
##########
@@ -488,19 +488,70 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
LOG_DEBUG("Process message chunk (chunkId: " << chunkId << ", uuid: " << uuid
<< ", messageId: " << messageId << ") of "
<< payload.readableBytes() << " bytes");
-
Lock lock(chunkProcessMutex_);
-
+ // For non-last chunks, increase available permits immediately since they don't occupy receiver queue.
+ if (chunkId != metadata.num_chunks_from_msg() - 1) {
+ increaseAvailablePermits(cnx);
+ }
// Lazy task scheduling to expire incomplete chunk message
bool expected = false;
if (expireTimeOfIncompleteChunkedMessageMs_ > 0 &&
expireChunkMessageTaskScheduled_.compare_exchange_strong(expected, true)) {
triggerCheckExpiredChunkedTimer();
}
+ // Part 1: chunkId == 0, this is the first chunk of a chunked message.
+ // If a previous incomplete context with the same uuid exists, it means either:
+ // a) Message redeliver: the first chunk's messageId matches one of the cached chunk messageIds.
+ // In this case, the old context is simply removed and a new context is created to restart
+ // assembling from scratch. No ack is needed since the old chunks will be redelivered.
+ // b) Corrupted chunk message: the first chunk's messageId does NOT match any cached chunk messageId,
+ // meaning a new producer sent a message with the same uuid. In this case, ack the old cached
+ // chunks to avoid ack holes, then remove the old context and create a new one.
+ // After handling the old context, check maxPendingChunkedMessage limit and create a new context.
auto it = chunkedMessageCache_.find(uuid);
-
- if (chunkId == 0 && it == chunkedMessageCache_.end()) {
+ if (chunkId == 0) {
+ // Handle ack hole when receiving duplicated first chunk.
+ // For example (message redeliver):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // For example (corrupted chunk message):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ if (it != chunkedMessageCache_.end()) {
+ auto& existingCtx = it->second;
+ bool isCorruptedChunkMessage = true;
+ for (const MessageId& cachedMsgId : existingCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunkMessage = false;
+ break;
Review Comment:
Comparing only ledgerId and entryId is intentional here. Chunk messages are never batched (batchIndex is always -1), and partition is constant within the same consumer. So ledgerId + entryId is sufficient to uniquely identify a chunk entry.
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
}
+ // Part 3: chunkId does not match the expected next chunk ID (out-of-order).
+ // Two sub-cases:
+ // a) chunkId <= lastChunkedMessageId: duplicated chunk caused by redeliver or corruption.
+ // Filter and ack the duplicated chunk if it's corrupted, then discard it.
+ // b) chunkId > lastChunkedMessageId + 1: gap detected, the chunked message is corrupted.
+ // Remove the context and ack the current chunk if it has expired.
+ auto& chunkedMsgCtx = it->second;
+ if (!chunkedMsgCtx.validateChunkId(chunkId)) {
+ const int lastChunkedMessageId = static_cast<int>(chunkedMsgCtx.getChunkedMessageIds().size()) - 1;
+ // For example (duplicated chunk):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ // Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
+ if (chunkId <= lastChunkedMessageId) {
+ bool isCorruptedChunk = true;
+ for (const MessageId& cachedMsgId : chunkedMsgCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunk = false;
+ break;
Review Comment:
Comparing only ledgerId and entryId is intentional here. Chunk messages are never batched (batchIndex is always -1), and partition is constant within the same consumer. So ledgerId + entryId is sufficient to uniquely identify a chunk entry.
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3339459897
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
}
+ // Part 3: chunkId does not match the expected next chunk ID (out-of-order).
+ // Two sub-cases:
+ // a) chunkId <= lastChunkedMessageId: duplicated chunk caused by redeliver or corruption.
+ // Filter and ack the duplicated chunk if it's corrupted, then discard it.
+ // b) chunkId > lastChunkedMessageId + 1: gap detected, the chunked message is corrupted.
+ // Remove the context and ack the current chunk if it has expired.
+ auto& chunkedMsgCtx = it->second;
+ if (!chunkedMsgCtx.validateChunkId(chunkId)) {
+ const int lastChunkedMessageId = static_cast<int>(chunkedMsgCtx.getChunkedMessageIds().size()) - 1;
+ // For example (duplicated chunk):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ // Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
+ if (chunkId <= lastChunkedMessageId) {
+ bool isCorruptedChunk = true;
+ for (const MessageId& cachedMsgId : chunkedMsgCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunk = false;
+ break;
+ }
+ }
+ LOG_WARN("Received a duplicated chunk message (uuid: "
+ << uuid << " chunkId: " << chunkId << ", lastChunkedMessageId: " << lastChunkedMessageId
+ << ", messageId: " << messageId << ")");
+ if (isCorruptedChunk) {
+ LOG_INFO("Acking corrupted duplicated chunk to avoid ack hole, uuid: "
+ << uuid << ", messageId: " << messageId);
+ acknowledgeAsync(messageId, [uuid, messageId](Result result) {
+ if (result != ResultOk) {
+ LOG_WARN("Failed to acknowledge duplicated chunk, uuid: "
+ << uuid << ", messageId: " << messageId);
+ }
+ });
+ }
+ lock.unlock();
+ return {};
+ }
+ // chunkId > lastChunkedMessageId + 1, the chunked message is corrupted.
+ LOG_WARN("Received unexpected chunk (uuid: "
+ << uuid << " chunkId: " << chunkId << ", lastChunkedMessageId: " << lastChunkedMessageId
+ << ", messageId: " << messageId << ")");
+ chunkedMessageCache_.remove(uuid);
+ lock.unlock();
Review Comment:
Fixed. Same permit leak issue — added the same guard here.
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3558406933
##########
lib/ConsumerImpl.cc:
##########
@@ -512,47 +563,117 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
Review Comment:
When a non-first chunk arrives without a cached context, the current logic simply logs the event and returns, leaving the incoming chunk neither acknowledged nor tracked. This may result in ack holes.
The releated java client logic is: https://github.com/apache/pulsar/blob/e8df873ddfde4d8b5a38a49e4b8765c693d65ab9/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1693-L1699
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#discussion_r3338289300
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
Review Comment:
With permits now only being returned up-front for non-last chunks, discarding a *last* chunk due to a missing chunk context (seek/cache eviction) will leak a permit and can eventually stall consumption.
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
}
+ // Part 3: chunkId does not match the expected next chunk ID (out-of-order).
+ // Two sub-cases:
+ // a) chunkId <= lastChunkedMessageId: duplicated chunk caused by redeliver or corruption.
+ // Filter and ack the duplicated chunk if it's corrupted, then discard it.
+ // b) chunkId > lastChunkedMessageId + 1: gap detected, the chunked message is corrupted.
+ // Remove the context and ack the current chunk if it has expired.
+ auto& chunkedMsgCtx = it->second;
+ if (!chunkedMsgCtx.validateChunkId(chunkId)) {
+ const int lastChunkedMessageId = static_cast<int>(chunkedMsgCtx.getChunkedMessageIds().size()) - 1;
+ // For example (duplicated chunk):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ // Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
+ if (chunkId <= lastChunkedMessageId) {
+ bool isCorruptedChunk = true;
+ for (const MessageId& cachedMsgId : chunkedMsgCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunk = false;
+ break;
+ }
+ }
+ LOG_WARN("Received a duplicated chunk message (uuid: "
+ << uuid << " chunkId: " << chunkId << ", lastChunkedMessageId: " << lastChunkedMessageId
+ << ", messageId: " << messageId << ")");
+ if (isCorruptedChunk) {
+ LOG_INFO("Acking corrupted duplicated chunk to avoid ack hole, uuid: "
+ << uuid << ", messageId: " << messageId);
+ acknowledgeAsync(messageId, [uuid, messageId](Result result) {
+ if (result != ResultOk) {
+ LOG_WARN("Failed to acknowledge duplicated chunk, uuid: "
+ << uuid << ", messageId: " << messageId);
+ }
+ });
+ }
+ lock.unlock();
+ return {};
+ }
+ // chunkId > lastChunkedMessageId + 1, the chunked message is corrupted.
+ LOG_WARN("Received unexpected chunk (uuid: "
+ << uuid << " chunkId: " << chunkId << ", lastChunkedMessageId: " << lastChunkedMessageId
+ << ", messageId: " << messageId << ")");
+ chunkedMessageCache_.remove(uuid);
+ lock.unlock();
Review Comment:
In the gap/out-of-order path, a *last* chunk can be discarded without returning its permit (permits are currently only returned early for non-last chunks). This can leak permits and stop further message delivery.
##########
tests/MessageChunkingTest.cc:
##########
@@ -96,6 +96,18 @@ class MessageChunkingTest : public ::testing::TestWithParam<CompressionType> {
std::string MessageChunkingTest::largeMessage = createLargeMessage();
+// Helper function: send a single chunk message
+static void sendSingleChunk(Producer& producer, const std::string& uuid, int chunkId, int totalChunks) {
+ std::string content = "chunk-" + uuid + "-" + std::to_string(chunkId) + "|";
+ auto msg = MessageBuilder().setContent(content).build();
+ auto& metadata = PulsarFriend::getMessageMetadata(msg);
+ metadata.set_num_chunks_from_msg(totalChunks);
+ metadata.set_chunk_id(chunkId);
+ metadata.set_uuid(uuid);
+ metadata.set_total_chunk_msg_size(100);
+ producer.send(msg);
Review Comment:
`sendSingleChunk` calls the deprecated `Producer::send(const Message&)` overload and ignores the returned `Result`, which makes test failures less actionable (a send failure can later look like a receive timeout).
##########
lib/ConsumerImpl.cc:
##########
@@ -488,19 +488,70 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
LOG_DEBUG("Process message chunk (chunkId: " << chunkId << ", uuid: " << uuid
<< ", messageId: " << messageId << ") of "
<< payload.readableBytes() << " bytes");
-
Lock lock(chunkProcessMutex_);
-
+ // For non-last chunks, increase available permits immediately since they don't occupy receiver queue.
+ if (chunkId != metadata.num_chunks_from_msg() - 1) {
+ increaseAvailablePermits(cnx);
+ }
// Lazy task scheduling to expire incomplete chunk message
bool expected = false;
if (expireTimeOfIncompleteChunkedMessageMs_ > 0 &&
expireChunkMessageTaskScheduled_.compare_exchange_strong(expected, true)) {
triggerCheckExpiredChunkedTimer();
}
+ // Part 1: chunkId == 0, this is the first chunk of a chunked message.
+ // If a previous incomplete context with the same uuid exists, it means either:
+ // a) Message redeliver: the first chunk's messageId matches one of the cached chunk messageIds.
+ // In this case, the old context is simply removed and a new context is created to restart
+ // assembling from scratch. No ack is needed since the old chunks will be redelivered.
+ // b) Corrupted chunk message: the first chunk's messageId does NOT match any cached chunk messageId,
+ // meaning a new producer sent a message with the same uuid. In this case, ack the old cached
+ // chunks to avoid ack holes, then remove the old context and create a new one.
+ // After handling the old context, check maxPendingChunkedMessage limit and create a new context.
auto it = chunkedMessageCache_.find(uuid);
-
- if (chunkId == 0 && it == chunkedMessageCache_.end()) {
+ if (chunkId == 0) {
+ // Handle ack hole when receiving duplicated first chunk.
+ // For example (message redeliver):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // For example (corrupted chunk message):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ if (it != chunkedMessageCache_.end()) {
+ auto& existingCtx = it->second;
+ bool isCorruptedChunkMessage = true;
+ for (const MessageId& cachedMsgId : existingCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunkMessage = false;
+ break;
Review Comment:
Corruption vs redelivery detection should use full `MessageId` equality (`operator==` includes ledgerId/entryId/batchIndex/partition). Comparing only ledgerId/entryId can misclassify IDs and lead to incorrect acking/cleanup decisions.
##########
lib/ConsumerImpl.cc:
##########
@@ -512,45 +563,104 @@ optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& pay
}
it = chunkedMessageCache_.putIfAbsent(
uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+ it->second.appendChunk(messageId, payload);
+ lock.unlock();
+ return {};
}
- auto& chunkedMsgCtx = it->second;
- if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+ // Part 2: chunkId != 0 but chunk context not found in cache.
+ // This happens when the first chunk was not received (e.g., consumer used seek() or started
+ // consuming from a specific message position that falls in the middle of a chunked message,
+ // or the context was evicted due to maxPendingChunkedMessage limit).
+ // In this case, the chunk message cannot be assembled, so just discard it.
+ if (it == chunkedMessageCache_.end()) {
auto startMessageId = getStartMessageId();
if (!config_.isStartMessageIdInclusive() && startMessageId &&
startMessageId->ledgerId() == messageId.ledgerId() &&
startMessageId->entryId() == messageId.entryId()) {
- // When the start message id is not inclusive, the last chunk of the previous chunked message will
- // be delivered, which is expected and we only need to filter it out.
- chunkedMessageCache_.remove(uuid);
LOG_INFO("Filtered the chunked message before the start message id (uuid: "
<< uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- } else if (it == chunkedMessageCache_.end()) {
+ } else {
LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
<< ", messageId: " << messageId << ")");
- } else {
- LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
- << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
- chunkedMessageCache_.remove(uuid);
}
lock.unlock();
- increaseAvailablePermits(cnx);
- trackMessage(messageId);
return {};
}
+ // Part 3: chunkId does not match the expected next chunk ID (out-of-order).
+ // Two sub-cases:
+ // a) chunkId <= lastChunkedMessageId: duplicated chunk caused by redeliver or corruption.
+ // Filter and ack the duplicated chunk if it's corrupted, then discard it.
+ // b) chunkId > lastChunkedMessageId + 1: gap detected, the chunked message is corrupted.
+ // Remove the context and ack the current chunk if it has expired.
+ auto& chunkedMsgCtx = it->second;
+ if (!chunkedMsgCtx.validateChunkId(chunkId)) {
+ const int lastChunkedMessageId = static_cast<int>(chunkedMsgCtx.getChunkedMessageIds().size()) - 1;
+ // For example (duplicated chunk):
+ // Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+ // Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+ // Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
+ // Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+ // Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+ // Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
+ if (chunkId <= lastChunkedMessageId) {
+ bool isCorruptedChunk = true;
+ for (const MessageId& cachedMsgId : chunkedMsgCtx.getChunkedMessageIds()) {
+ if (cachedMsgId.ledgerId() == messageId.ledgerId() &&
+ cachedMsgId.entryId() == messageId.entryId()) {
+ isCorruptedChunk = false;
+ break;
Review Comment:
Same issue here: use `cachedMsgId == messageId` to decide whether a duplicated chunk is a redelivery. This prevents treating a legitimate redelivery as corruption if other MessageId fields differ.
--
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] [improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks [pulsar-client-cpp]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #587:
URL: https://github.com/apache/pulsar-client-cpp/pull/587#issuecomment-4601182099
@geniusjoe You can create the same PR in your own fork, so you don't need to wait approval from committers to run workflow
--
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