You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2023/08/31 10:50:01 UTC
[pulsar] branch branch-3.1 updated (f6d20716e7a -> 4503540c80f)
This is an automated email from the ASF dual-hosted git repository.
zike pushed a change to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from f6d20716e7a [fix][broker] Make sure all inflight writes have finished before completion of compaction (#21067)
new 906fbbacc53 [fix][client] Fix consumer can't consume resent chunked messages (#21070)
new 4503540c80f [fix][client] Fix cannot retry chunk messages and send to DLQ (#21048)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/client/api/DeadLetterTopicTest.java | 48 +++++++++++++++++----
.../pulsar/client/impl/MessageChunkingTest.java | 41 ++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 18 +++++++-
.../pulsar/client/impl/MessageIdAdvUtils.java | 3 ++
.../apache/pulsar/client/impl/ProducerImpl.java | 6 +++
.../pulsar/client/impl/UnAckedMessageTracker.java | 7 +++-
.../client/impl/UnAckedMessageTrackerTest.java | 49 +++++++++++++++++++++-
7 files changed, 159 insertions(+), 13 deletions(-)
[pulsar] 01/02: [fix][client] Fix consumer can't consume resent chunked messages (#21070)
Posted by zi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zike pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 906fbbacc537e23adb2052f2b225eca247719793
Author: Zike Yang <zi...@apache.org>
AuthorDate: Wed Aug 30 00:23:52 2023 +0800
[fix][client] Fix consumer can't consume resent chunked messages (#21070)
### 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:
https://github.com/apache/pulsar/blob/44a055b8a55078bcf93f4904991598541aa6c1ee/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1436-L1482
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.
(cherry picked from commit eb2e3a258b971cfeeb22f1cec254cafb49d0ae40)
---
.../pulsar/client/impl/MessageChunkingTest.java | 41 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 12 ++++++-
2 files changed, 52 insertions(+), 1 deletion(-)
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 2797d66f7fe..dffa0035248 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,47 @@ public class MessageChunkingTest extends ProducerConsumerBase {
assertNull(consumer.receive(5, TimeUnit.SECONDS));
}
+ @Test
+ public void testResendChunkMessages() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+ final String topicName = "persistent://my-property/my-ns/testResendChunkMessages";
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName("my-subscriber-name")
+ .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, "1", 0, 3); // This is for testing the interwoven chunked message
+ sendSingleChunk(producer, "1", 1, 3);
+ sendSingleChunk(producer, "1", 0, 3); // Resending the UUID-1 chunked message
+
+ 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);
+
+ sendSingleChunk(producer, "1", 1, 3);
+ sendSingleChunk(producer, "1", 2, 3);
+
+ receivedMsg = consumer.receive(5, TimeUnit.SECONDS);
+ assertEquals(receivedMsg.getValue(), "chunk-1-0|chunk-1-1|chunk-1-2|");
+ consumer.acknowledge(receivedMsg);
+ }
+
/**
* Validate that chunking is not supported with batching and non-persistent topic
*
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 a929fe9aa6b..e914245ffa4 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
@@ -1433,7 +1433,17 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.get(msgMetadata.getUuid());
- if (msgMetadata.getChunkId() == 0 && chunkedMsgCtx == null) {
+ 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();
[pulsar] 02/02: [fix][client] Fix cannot retry chunk messages and send to DLQ (#21048)
Posted by zi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zike pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4503540c80fa07eaf7a3701948e3e6b7b1135e3d
Author: Zike Yang <zi...@apache.org>
AuthorDate: Thu Aug 31 07:44:34 2023 +0800
[fix][client] Fix cannot retry chunk messages and send to DLQ (#21048)
(cherry picked from commit 99e3fea2e3adecb3346d1fc8e5422a3fde4236e3)
---
.../pulsar/client/api/DeadLetterTopicTest.java | 48 +++++++++++++++++----
.../apache/pulsar/client/impl/ConsumerImpl.java | 6 ++-
.../pulsar/client/impl/MessageIdAdvUtils.java | 3 ++
.../apache/pulsar/client/impl/ProducerImpl.java | 6 +++
.../pulsar/client/impl/UnAckedMessageTracker.java | 7 +++-
.../client/impl/UnAckedMessageTrackerTest.java | 49 +++++++++++++++++++++-
6 files changed, 107 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
index 5e3731fbf24..2a0cb3187d2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
@@ -29,6 +29,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -46,9 +47,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
-@Test(groups = "flaky")
+@Test(groups = "broker-impl")
public class DeadLetterTopicTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(DeadLetterTopicTest.class);
@@ -56,6 +58,7 @@ public class DeadLetterTopicTest extends ProducerConsumerBase {
@BeforeMethod(alwaysRun = true)
@Override
protected void setup() throws Exception {
+ this.conf.setMaxMessageSize(5 * 1024);
super.internalSetup();
super.producerBaseSetup();
}
@@ -66,6 +69,15 @@ public class DeadLetterTopicTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ private String createMessagePayload(int size) {
+ StringBuilder str = new StringBuilder();
+ Random rand = new Random();
+ for (int i = 0; i < size; i++) {
+ str.append(rand.nextInt(10));
+ }
+ return str.toString();
+ }
+
@Test
public void testDeadLetterTopicWithMessageKey() throws Exception {
final String topic = "persistent://my-property/my-ns/dead-letter-topic";
@@ -125,9 +137,13 @@ public class DeadLetterTopicTest extends ProducerConsumerBase {
consumer.close();
}
+ @DataProvider(name = "produceLargeMessages")
+ public Object[][] produceLargeMessages() {
+ return new Object[][] { { false }, { true } };
+ }
- @Test(groups = "quarantine")
- public void testDeadLetterTopic() throws Exception {
+ @Test(dataProvider = "produceLargeMessages")
+ public void testDeadLetterTopic(boolean produceLargeMessages) throws Exception {
final String topic = "persistent://my-property/my-ns/dead-letter-topic";
final int maxRedeliveryCount = 2;
@@ -154,28 +170,44 @@ public class DeadLetterTopicTest extends ProducerConsumerBase {
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topic)
+ .enableChunking(produceLargeMessages)
+ .enableBatching(!produceLargeMessages)
.create();
+ Map<Integer, String> messageContent = new HashMap<>();
+
for (int i = 0; i < sendMessages; i++) {
- producer.send(String.format("Hello Pulsar [%d]", i).getBytes());
+ String data;
+ if (!produceLargeMessages) {
+ data = String.format("Hello Pulsar [%d]", i);
+ } else {
+ data = createMessagePayload(1024 * 10);
+ }
+ producer.newMessage().key(String.valueOf(i)).value(data.getBytes()).send();
+ messageContent.put(i, data);
}
producer.close();
int totalReceived = 0;
do {
- Message<byte[]> message = consumer.receive();
- log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
+ Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertNotNull(message, "The consumer should be able to receive messages.");
+ log.info("consumer received message : {}", message.getMessageId());
totalReceived++;
} while (totalReceived < sendMessages * (maxRedeliveryCount + 1));
int totalInDeadLetter = 0;
do {
- Message message = deadLetterConsumer.receive();
- log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
+ Message message = deadLetterConsumer.receive(5, TimeUnit.SECONDS);
+ assertNotNull(message, "the deadLetterConsumer should receive messages.");
+ assertEquals(new String(message.getData()), messageContent.get(Integer.parseInt(message.getKey())));
+ messageContent.remove(Integer.parseInt(message.getKey()));
+ log.info("dead letter consumer received message : {}", message.getMessageId());
deadLetterConsumer.acknowledge(message);
totalInDeadLetter++;
} while (totalInDeadLetter < sendMessages);
+ assertTrue(messageContent.isEmpty());
deadLetterConsumer.close();
consumer.close();
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 e914245ffa4..72932467d7f 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
@@ -613,6 +613,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
retryLetterProducer = client.newProducer(Schema.AUTO_PRODUCE_BYTES(schema))
.topic(this.deadLetterPolicy.getRetryLetterTopic())
.enableBatching(false)
+ .enableChunking(true)
.blockIfQueueFull(false)
.create();
}
@@ -1460,7 +1461,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
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,
+ log.info("[{}] [{}] Received unexpected chunk messageId {}, last-chunk-id = {}, chunkId = {}", topic,
+ subscription, msgId,
(chunkedMsgCtx != null ? chunkedMsgCtx.lastChunkedMessageId : null), msgMetadata.getChunkId());
if (chunkedMsgCtx != null) {
if (chunkedMsgCtx.chunkedMsgBuffer != null) {
@@ -2104,6 +2106,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
.initialSubscriptionName(this.deadLetterPolicy.getInitialSubscriptionName())
.topic(this.deadLetterPolicy.getDeadLetterTopic())
.blockIfQueueFull(false)
+ .enableBatching(false)
+ .enableChunking(true)
.createAsync();
}
} finally {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java
index c8b18524ec0..a0d1446ba3d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java
@@ -64,6 +64,9 @@ public class MessageIdAdvUtils {
}
static MessageIdAdv discardBatch(MessageId messageId) {
+ if (messageId instanceof ChunkMessageIdImpl) {
+ return (MessageIdAdv) messageId;
+ }
MessageIdAdv msgId = (MessageIdAdv) messageId;
return new MessageIdImpl(msgId.getLedgerId(), msgId.getEntryId(), msgId.getPartitionIndex());
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 267b06649d7..8a30190b3a2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -695,6 +695,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
op = OpSendMsg.create(msg, null, sequenceId, callback);
final MessageMetadata finalMsgMetadata = msgMetadata;
op.rePopulate = () -> {
+ if (msgMetadata.hasChunkId()) {
+ // The message metadata is shared between all chunks in a large message
+ // We need to reset the chunk id for each call of this method
+ // It's safe to do that because there is only 1 thread to manipulate this message metadata
+ finalMsgMetadata.setChunkId(chunkId);
+ }
op.cmd = sendMessage(producerId, sequenceId, numMessages, messageId, finalMsgMetadata,
encryptedPayload);
};
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
index 69f86a1a89f..20ec9c3d99a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
@@ -138,8 +138,11 @@ public class UnAckedMessageTracker implements Closeable {
if (!headPartition.isEmpty()) {
log.info("[{}] {} messages will be re-delivered", consumerBase, headPartition.size());
headPartition.forEach(messageId -> {
- addChunkedMessageIdsAndRemoveFromSequenceMap(messageId, messageIds, consumerBase);
- messageIds.add(messageId);
+ if (messageId instanceof ChunkMessageIdImpl) {
+ addChunkedMessageIdsAndRemoveFromSequenceMap(messageId, messageIds, consumerBase);
+ } else {
+ messageIds.add(messageId);
+ }
messageIdPartitionMap.remove(messageId);
});
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/UnAckedMessageTrackerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/UnAckedMessageTrackerTest.java
index 4ccc514e8e7..f6c668703d9 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/UnAckedMessageTrackerTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/UnAckedMessageTrackerTest.java
@@ -29,12 +29,15 @@ import static org.mockito.Mockito.when;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
import io.netty.util.concurrent.DefaultThreadFactory;
-
+import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.awaitility.Awaitility;
import org.testng.annotations.Test;
public class UnAckedMessageTrackerTest {
@@ -77,4 +80,48 @@ public class UnAckedMessageTrackerTest {
timer.stop();
}
+ @Test
+ public void testTrackChunkedMessageId() {
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ Timer timer = new HashedWheelTimer(new DefaultThreadFactory("pulsar-timer", Thread.currentThread().isDaemon()),
+ 1, TimeUnit.MILLISECONDS);
+ when(client.timer()).thenReturn(timer);
+
+ ConsumerBase<byte[]> consumer = mock(ConsumerBase.class);
+ doNothing().when(consumer).onAckTimeoutSend(any());
+ doNothing().when(consumer).redeliverUnacknowledgedMessages(any());
+ ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
+ conf.setAckTimeoutMillis(1000);
+ conf.setTickDurationMillis(1000);
+ UnAckedMessageTracker tracker = new UnAckedMessageTracker(client, consumer, conf);
+
+ assertTrue(tracker.isEmpty());
+ assertEquals(tracker.size(), 0);
+
+ // Build chunked message ID
+ MessageIdImpl[] chunkMsgIds = new MessageIdImpl[5];
+ for (int i = 0; i < 5; i++) {
+ chunkMsgIds[i] = new MessageIdImpl(1L, i, -1);
+ }
+ ChunkMessageIdImpl chunkedMessageId =
+ new ChunkMessageIdImpl(chunkMsgIds[0], chunkMsgIds[chunkMsgIds.length - 1]);
+
+ consumer.unAckedChunkedMessageIdSequenceMap =
+ ConcurrentOpenHashMap.<MessageIdAdv, MessageIdImpl[]>newBuilder().build();
+ consumer.unAckedChunkedMessageIdSequenceMap.put(chunkedMessageId, chunkMsgIds);
+
+ // Redeliver chunked message
+ tracker.add(chunkedMessageId);
+
+ Awaitility.await()
+ .pollInterval(Duration.ofMillis(200))
+ .atMost(Duration.ofSeconds(3))
+ .untilAsserted(() -> assertEquals(tracker.size(), 0));
+
+ // Assert that all chunk message ID are removed from unAckedChunkedMessageIdSequenceMap
+ assertEquals(consumer.unAckedChunkedMessageIdSequenceMap.size(), 0);
+
+ timer.stop();
+ }
+
}