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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
