[fix][client] Fix consumer can't consume resent chunked messages

[RobertIndie]

Motivation

Current, when the producer resend the chunked message like this:

• M1: UUID: 0, ChunkID: 0
• M2: UUID: 0, ChunkID: 0 // Resend the first chunk
• M3: UUID: 0, ChunkID: 1

When the consumer received the M2, it will find that it's already tracking the UUID:0 chunked messages, and will then discard the message M1 and M2. This will lead to unable to consume the whole chunked message even though it's already persisted in the Pulsar topic.

Here is the code logic:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Lines 1436 to 1482 in 44a055b

	if (msgMetadata.getChunkId() == 0) {
	if (chunkedMsgCtx != null) {
	// The first chunk of a new chunked-message received before receiving other chunks of previous
	// chunked-message
	// so, remove previous chunked-message from map and release buffer
	if (chunkedMsgCtx.chunkedMsgBuffer != null) {
	ReferenceCountUtil.safeRelease(chunkedMsgCtx.chunkedMsgBuffer);
	}
	chunkedMsgCtx.recycle();
	chunkedMessagesMap.remove(msgMetadata.getUuid());
	}
	pendingChunkedMessageCount++;
	if (maxPendingChunkedMessage > 0 && pendingChunkedMessageCount > maxPendingChunkedMessage) {
	removeOldestPendingChunkedMessage();
	}
	int totalChunks = msgMetadata.getNumChunksFromMsg();
	ByteBuf chunkedMsgBuffer = PulsarByteBufAllocator.DEFAULT.buffer(msgMetadata.getTotalChunkMsgSize(),
	msgMetadata.getTotalChunkMsgSize());
	chunkedMsgCtx = chunkedMessagesMap.computeIfAbsent(msgMetadata.getUuid(),
	(key) -> ChunkedMessageCtx.get(totalChunks, chunkedMsgBuffer));
	pendingChunkedMessageUuidQueue.add(msgMetadata.getUuid());
	}
	// discard message if chunk is out-of-order
	if (chunkedMsgCtx == null || chunkedMsgCtx.chunkedMsgBuffer == null
	|| msgMetadata.getChunkId() != (chunkedMsgCtx.lastChunkedMessageId + 1)) {
	// means we lost the first chunk: should never happen
	log.info("Received unexpected chunk messageId {}, last-chunk-id{}, chunkId = {}", msgId,
	(chunkedMsgCtx != null ? chunkedMsgCtx.lastChunkedMessageId : null), msgMetadata.getChunkId());
	if (chunkedMsgCtx != null) {
	if (chunkedMsgCtx.chunkedMsgBuffer != null) {
	ReferenceCountUtil.safeRelease(chunkedMsgCtx.chunkedMsgBuffer);
	}
	chunkedMsgCtx.recycle();
	}
	chunkedMessagesMap.remove(msgMetadata.getUuid());
	compressedPayload.release();
	increaseAvailablePermits(cnx);
	if (expireTimeOfIncompleteChunkedMessageMillis > 0
	&& System.currentTimeMillis() > (msgMetadata.getPublishTime()
	+ expireTimeOfIncompleteChunkedMessageMillis)) {
	doAcknowledge(msgId, AckType.Individual, Collections.emptyMap(), null);
	} else {
	trackMessage(msgId);
	}
	return null;
	}

The bug can be easily reproduced using the testcase testResendChunkMessages introduced by this PR.

Modifications

• When receiving the new duplicated first chunk of a chunked message, the consumer discard the current chunked message context and create a new context to track the following messages. For the case mentioned in Motivation, the M1 will be released and the consumer will assemble M2 and M3 as the chunked message.

Verifying this change

This change added tests.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[github-actions]

@RobertIndie Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->