You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/09/05 03:54:05 UTC
[pulsar] branch branch-3.0 updated: [fix][client] Avoid ack hole for chunk message (#21101)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new f3b50d6c114 [fix][client] Avoid ack hole for chunk message (#21101)
f3b50d6c114 is described below
commit f3b50d6c114f12ec8e65fa0234f8ab796ed5cf33
Author: Xiangying Meng <55...@users.noreply.github.com>
AuthorDate: Mon Sep 4 08:50:49 2023 +0800
[fix][client] Avoid ack hole for chunk message (#21101)
## Motivation
Handle ack hole case:
For example:
```markdown
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
```
Consumer ack chunk message via ChunkMessageIdImpl that consists of all the chunks in this chunk
message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the
ChunkMessageIdImpl, so we should process it here.
## Modification
Ack chunk-1 and chunk-2.
---
.../pulsar/client/impl/MessageChunkingTest.java | 33 ++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 46 ++++++++++++++++++----
2 files changed, 72 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
index dffa0035248..f266afd8a2e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
@@ -356,6 +356,38 @@ public class MessageChunkingTest extends ProducerConsumerBase {
assertNull(consumer.receive(5, TimeUnit.SECONDS));
}
+ @Test
+ public void testResendChunkMessagesWithoutAckHole() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+ final String topicName = "persistent://my-property/my-ns/testResendChunkMessagesWithoutAckHole";
+ final String subName = "my-subscriber-name";
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName(subName)
+ .maxPendingChunkedMessage(10)
+ .autoAckOldestChunkedMessageOnQueueFull(true)
+ .subscribe();
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .chunkMaxMessageSize(100)
+ .enableChunking(true)
+ .enableBatching(false)
+ .create();
+
+ sendSingleChunk(producer, "0", 0, 2);
+
+ sendSingleChunk(producer, "0", 0, 2); // Resending the first chunk
+ sendSingleChunk(producer, "0", 1, 2);
+
+ Message<String> receivedMsg = consumer.receive(5, TimeUnit.SECONDS);
+ assertEquals(receivedMsg.getValue(), "chunk-0-0|chunk-0-1|");
+ consumer.acknowledge(receivedMsg);
+ assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName)
+ .getNonContiguousDeletedMessagesRanges(), 0);
+ }
+
@Test
public void testResendChunkMessages() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -395,6 +427,7 @@ public class MessageChunkingTest extends ProducerConsumerBase {
receivedMsg = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(receivedMsg.getValue(), "chunk-1-0|chunk-1-1|chunk-1-2|");
consumer.acknowledge(receivedMsg);
+ Assert.assertEquals(((ConsumerImpl<String>) consumer).getAvailablePermits(), 8);
}
/**
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 8263e7f3198..3a542e66104 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1413,7 +1413,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
private ByteBuf processMessageChunk(ByteBuf compressedPayload, MessageMetadata msgMetadata, MessageIdImpl msgId,
MessageIdData messageId, ClientCnx cnx) {
-
+ if (msgMetadata.getChunkId() != (msgMetadata.getNumChunksFromMsg() - 1)) {
+ increaseAvailablePermits(cnx);
+ }
// Lazy task scheduling to expire incomplete chunk message
if (expireTimeOfIncompleteChunkedMessageMillis > 0 && expireChunkMessageTaskScheduled.compareAndSet(false,
true)) {
@@ -1429,6 +1431,37 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (msgMetadata.getChunkId() == 0) {
if (chunkedMsgCtx != null) {
+ // Handle ack hole case when receive duplicated chunks.
+ // There are two situation that receives chunks with the same sequence ID and chunk ID.
+ // Situation 1 - Message redeliver:
+ // For example:
+ // 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
+ // In this case, chunk-3 and chunk-4 have the same msgID with chunk-1 and chunk-2.
+ // This may be caused by message redeliver, we can't ack any chunk in this case here.
+ // Situation 2 - Corrupted chunk message
+ // For example:
+ // 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
+ // In this case, all the chunks with different msgIDs and are persistent in the topic.
+ // But Chunk-1 and Chunk-2 belong to a corrupted chunk message that must be skipped since
+ // they will not be delivered to end users. So we should ack them here to avoid ack hole.
+ boolean isCorruptedChunkMessageDetected = Arrays.stream(chunkedMsgCtx.chunkedMessageIds)
+ .noneMatch(messageId1 -> messageId1 != null && messageId1.ledgerId == messageId.getLedgerId()
+ && messageId1.entryId == messageId.getEntryId());
+ if (isCorruptedChunkMessageDetected) {
+ Arrays.stream(chunkedMsgCtx.chunkedMessageIds).forEach(messageId1 -> {
+ if (messageId1 != null) {
+ doAcknowledge(messageId1, AckType.Individual, Collections.emptyMap(), 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
@@ -1468,11 +1501,12 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
msgMetadata.getProducerName(), msgId, chunkedMsgCtx.lastChunkedMessageId,
msgMetadata.getChunkId(), msgMetadata.getSequenceId());
compressedPayload.release();
- increaseAvailablePermits(cnx);
- boolean repeatedlyReceived = Arrays.stream(chunkedMsgCtx.chunkedMessageIds)
- .anyMatch(messageId1 -> messageId1 != null && messageId1.ledgerId == messageId.getLedgerId()
+ // Just like the above logic of receiving the first chunk again. We only ack this chunk in the message
+ // duplication case.
+ boolean isDuplicatedChunk = Arrays.stream(chunkedMsgCtx.chunkedMessageIds)
+ .noneMatch(messageId1 -> messageId1 != null && messageId1.ledgerId == messageId.getLedgerId()
&& messageId1.entryId == messageId.getEntryId());
- if (!repeatedlyReceived) {
+ if (isDuplicatedChunk) {
doAcknowledge(msgId, AckType.Individual, Collections.emptyMap(), null);
}
return null;
@@ -1489,7 +1523,6 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
chunkedMessagesMap.remove(msgMetadata.getUuid());
compressedPayload.release();
- increaseAvailablePermits(cnx);
if (expireTimeOfIncompleteChunkedMessageMillis > 0
&& System.currentTimeMillis() > (msgMetadata.getPublishTime()
+ expireTimeOfIncompleteChunkedMessageMillis)) {
@@ -1508,7 +1541,6 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
// if final chunk is not received yet then release payload and return
if (msgMetadata.getChunkId() != (msgMetadata.getNumChunksFromMsg() - 1)) {
compressedPayload.release();
- increaseAvailablePermits(cnx);
return null;
}