You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 07:50:22 UTC
(pulsar) 04/17: [fix][client] Prevent chunk message ID leaks after seek (#26909)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e62775c773fffdfdd8f3b7cefa17832b04963499
Author: sinan liu <li...@gmail.com>
AuthorDate: Fri Oct 16 02:17:58 2026 +0800
[fix][client] Prevent chunk message ID leaks after seek (#26909)
(cherry picked from commit 25e10a53055977736c36d1f0e46e56a4e86dd749)
---
.../pulsar/client/impl/MessageChunkingTest.java | 89 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 9 ++-
2 files changed, 96 insertions(+), 2 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 0eb91b336f3f..2bc30ba9ca4f 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
@@ -50,8 +51,10 @@ import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
@@ -994,6 +997,92 @@ public class MessageChunkingTest extends ProducerConsumerBase {
log.info().attr("method", methodName).log("Exiting test");
}
+ @DataProvider(name = "seekInclusive")
+ public Object[][] seekInclusive() {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "seekInclusive")
+ public void testRepeatedSeekDoesNotRetainFilteredChunkMessageIds(boolean inclusive) throws Exception {
+ final String topicName = "persistent://my-property/my-ns/seek-chunk-message-ids";
+ final int totalMessages = 4;
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableChunking(true)
+ .enableBatching(false)
+ .chunkMaxMessageSize(100)
+ .create();
+ List<MessageId> messageIds = new ArrayList<>();
+ for (int i = 0; i < totalMessages; i++) {
+ MessageId messageId = producer.send(("message-" + i).repeat(30));
+ assertThat(messageId).isInstanceOf(ChunkMessageIdImpl.class);
+ messageIds.add(messageId);
+ }
+ MessageId markerId = producer.send("marker");
+ assertThat(markerId).isNotInstanceOf(ChunkMessageIdImpl.class);
+
+ // Subscribe after publishing so no chunk mappings exist before the first seek.
+ ConsumerBuilder<String> consumerBuilder = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName("my-sub")
+ .receiverQueueSize(32)
+ .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS);
+ if (inclusive) {
+ consumerBuilder.startMessageIdInclusive();
+ }
+ @Cleanup
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) consumerBuilder.subscribe();
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap).isEmpty();
+
+ for (int boundary = 0; boundary < totalMessages; boundary++) {
+ consumer.seek(messageIds.get(boundary));
+ for (int i = inclusive ? boundary : boundary + 1; i < totalMessages; i++) {
+ Message<String> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(message).isNotNull();
+ MessageIdAdv messageId = (MessageIdAdv) message.getMessageId();
+ assertThat(messageId).isEqualTo(messageIds.get(i));
+ assertThat(message.getValue()).isEqualTo(("message-" + i).repeat(30));
+ // Delivered messages must keep their chunk sequence until the application ACKs them.
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap).containsKey(messageId);
+ consumer.acknowledge(message);
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap).doesNotContainKey(messageId);
+ }
+ // Receiving the following marker proves the boundary has been assembled and filtered.
+ Message<String> marker = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(marker).isNotNull();
+ assertThat(marker.getMessageId()).isEqualTo(markerId);
+ consumer.acknowledge(marker);
+ }
+
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap)
+ .as("filtered seek boundaries must not accumulate chunk message ID sequences")
+ .isEmpty();
+
+ if (!inclusive) {
+ // A filtered redelivery must not remove the sequence of an earlier application-held message.
+ consumer.seek(MessageId.earliest);
+ Message<String> outstanding = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(outstanding).isNotNull();
+ MessageIdAdv outstandingId = (MessageIdAdv) outstanding.getMessageId();
+ assertThat(outstandingId).isEqualTo(messageIds.get(0));
+ consumer.seek(outstandingId);
+ for (int i = 1; i < totalMessages; i++) {
+ Message<String> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(message).isNotNull();
+ assertThat(message.getMessageId()).isEqualTo(messageIds.get(i));
+ consumer.acknowledge(message);
+ }
+ Message<String> marker = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(marker).isNotNull();
+ assertThat(marker.getMessageId()).isEqualTo(markerId);
+ consumer.acknowledge(marker);
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap).containsOnlyKeys(outstandingId);
+ consumer.acknowledge(outstanding);
+ assertThat(consumer.unAckedChunkedMessageIdSequenceMap).isEmpty();
+ }
+ }
+
@Test
public void testReaderChunkingConfiguration() throws Exception {
log.info().attr("method", methodName).log("Starting test");
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 8a52295d48f1..d9290cb19894 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
@@ -1518,6 +1518,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
// and return undecrypted payload
if (isMessageUndecryptable || (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch())) {
+ MessageIdImpl[] chunkedMessageIds = null;
if (isChunkedMessage) {
uncompressedPayload = processMessageChunk(uncompressedPayload, msgMetadata, msgId, messageId, cnx);
if (uncompressedPayload == null) {
@@ -1537,8 +1538,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
msgId = new ChunkMessageIdImpl(chunkedMsgCtx.chunkedMessageIds[0],
chunkedMsgCtx.chunkedMessageIds[chunkedMsgCtx.chunkedMessageIds.length - 1]);
}
- // add chunked messageId to unack-message tracker, and reduce pending-chunked-message count
- unAckedChunkedMessageIdSequenceMap.put(msgId, chunkedMsgCtx.chunkedMessageIds);
+ chunkedMessageIds = chunkedMsgCtx.chunkedMessageIds;
pendingChunkedMessageCount--;
// The completed message's uuid was added to pendingChunkedMessageUuidQueue when its
// first chunk arrived, but is only ever removed by the eviction/expiry paths. On the
@@ -1567,6 +1567,11 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return;
}
+ if (chunkedMessageIds != null) {
+ // Only register messages that pass filtering and can be acknowledged by the application.
+ unAckedChunkedMessageIdSequenceMap.put(msgId, chunkedMessageIds);
+ }
+
final MessageImpl<T> message =
newMessage(msgId, brokerEntryMetadata, msgMetadata, uncompressedPayload,
schema, redeliveryCount, consumerEpoch, isMessageUndecryptable);