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:34 UTC
(pulsar) 16/17: [fix][client] Make chunking inert on non-persistent topics, including the message-size checks (#26550)
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 d2d64a3daca2b96e3befa412275bd36ff628c7bd
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Fri Oct 16 10:27:37 2026 +0800
[fix][client] Make chunking inert on non-persistent topics, including the message-size checks (#26550)
(cherry picked from commit ca087eb44ab42e09579c964965daf41627e0e879)
---
.../pulsar/client/impl/MessageChunkingTest.java | 110 ++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 22 +++-
.../client/impl/BatchMessageContainerImplTest.java | 145 +++++++++++++++++++++
3 files changed, 273 insertions(+), 4 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 31120d22a984..d683be4b667e 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
@@ -61,6 +61,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.InvalidMessageException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SizeUnit;
@@ -713,6 +714,115 @@ public class MessageChunkingTest extends ProducerConsumerBase {
consumer.acknowledge(msg);
}
+ /**
+ * A non-persistent topic cannot carry chunked messages: the slicing block in
+ * {@code ProducerImpl.serializeAndSendMessage} is skipped for them, but {@code totalChunks} was computed
+ * from the payload size alone, so the send loop still ran once per "chunk" — publishing the whole payload
+ * that many times, and releasing the payload buffer more often than it was retained.
+ */
+ @Test
+ public void testLargeMessageOnNonPersistentTopicIsSentOnceWithoutChunking() throws Exception {
+ final String topicName = "non-persistent://my-property/my-ns/testNonPersistentChunking";
+ final String subName = "my-subscriber-name";
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName(subName)
+ .subscribe();
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ // small enough that the payload below would otherwise be split into several chunks
+ .chunkMaxMessageSize(100)
+ .enableChunking(true)
+ .enableBatching(false)
+ .create();
+
+ String payload = "a".repeat(500);
+ producer.newMessage().value(payload).send();
+
+ Message<String> received = consumer.receive(10, TimeUnit.SECONDS);
+ assertNotNull(received, "the message was not delivered");
+ assertEquals(received.getValue(), payload);
+
+ // Without the fix the same payload is published once per computed chunk, so more messages follow.
+ assertNull(consumer.receive(3, TimeUnit.SECONDS),
+ "the payload was published more than once on a non-persistent topic");
+ }
+
+ /**
+ * The chunking flag is inert on a non-persistent topic, so the message-size checks must behave as if it were
+ * off: a payload above the broker limit is rejected locally with {@link InvalidMessageException} instead of
+ * being sent to the broker as one oversized frame.
+ */
+ @Test
+ public void testLargeMessageOnNonPersistentTopicAboveBrokerLimitIsRejectedLocally() throws Exception {
+ final int maxMessageSize = 1024;
+ this.conf.setMaxMessageSize(maxMessageSize);
+ final String topicName = "non-persistent://my-property/my-ns/testNonPersistentChunkingAboveLimit";
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .enableChunking(true)
+ .enableBatching(false)
+ .sendTimeout(5, TimeUnit.SECONDS)
+ .create();
+
+ // Well above maxMessageSize plus the broker's frame padding: the payload cannot be sent as a single
+ // frame, so a local rejection is the only correct outcome.
+ byte[] payload = RandomUtils.nextBytes(maxMessageSize + Commands.MESSAGE_SIZE_FRAME_PADDING + 1024);
+ try {
+ producer.send(payload);
+ fail("an oversized message on a non-persistent topic must be rejected by the client");
+ } catch (InvalidMessageException expected) {
+ assertTrue(expected.getMessage().contains("exceeds"), expected.getMessage());
+ }
+ }
+
+ /**
+ * The final-size check runs on the complete frame, after the metadata has been added: a body within the limit
+ * whose metadata pushes the message over it must be rejected there on a non-persistent topic, since nothing will
+ * chunk it, and the producer must stay usable for the next valid message.
+ */
+ @Test
+ public void testNonPersistentMessageOverLimitOnlyWithMetadataIsRejectedLocally() throws Exception {
+ final int maxMessageSize = 1024;
+ final String topicName = "non-persistent://my-property/my-ns/testNonPersistentChunkingMetadataOverLimit";
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-subscriber-name")
+ .subscribe();
+
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topicName)
+ .enableChunking(true)
+ .enableBatching(false)
+ .sendTimeout(5, TimeUnit.SECONDS)
+ .create();
+ producer.getConnectionHandler().setMaxMessageSize(maxMessageSize);
+
+ // exactly at the limit, so the early check on the body passes and only the final check can reject it
+ byte[] payload = RandomUtils.nextBytes(maxMessageSize);
+ try {
+ producer.newMessage().value(payload).property("large-metadata", "y".repeat(400)).send();
+ fail("a message pushed over the limit by its metadata must be rejected by the client");
+ } catch (InvalidMessageException expected) {
+ assertTrue(expected.getMessage().contains("exceeds"), expected.getMessage());
+ }
+
+ byte[] valid = "valid".getBytes(StandardCharsets.UTF_8);
+ producer.send(valid);
+ Message<byte[]> received = consumer.receive(10, TimeUnit.SECONDS);
+ assertNotNull(received, "the valid message after the rejected one was not delivered");
+ assertEquals(received.getValue(), valid);
+ }
+
/**
* Verifies that pendingChunkedMessageUuidQueue does not leak entries as chunked messages
* complete normally. Each first chunk adds the message uuid to the queue; on completion the
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 d2d634bbccde..84e2042ea840 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
@@ -118,6 +118,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// Producer id, used to identify a producer within a single connection
protected final long producerId;
+ // A non-persistent topic cannot carry chunked messages, see isChunkingEnabled()
+ private final boolean persistentTopic;
// Variable is updated in a synchronized block
private volatile long msgIdGenerator;
@@ -207,6 +209,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
ProducerInterceptors interceptors, Optional<String> overrideProducerName) {
super(client, topic, conf, producerCreatedFuture, schema, interceptors);
this.producerId = client.newProducerId();
+ this.persistentTopic = TopicName.get(topic).isPersistent();
this.producerName = conf.getProducerName();
this.userProvidedProducerName = StringUtils.isNotBlank(producerName);
this.partitionIndex = partitionIndex;
@@ -607,7 +610,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// validate msg-size (For batching this will be check at the batch completion size)
int compressedSize = compressedPayload.readableBytes();
- if (compressedSize > getMaxMessageSize() && !this.conf.isChunkingEnabled()) {
+ if (compressedSize > getMaxMessageSize() && !isChunkingEnabled()) {
compressedPayload.release();
String compressedStr = conf.getCompressionType() != CompressionType.NONE
? ("compressed (" + conf.getCompressionType() + ")")
@@ -647,7 +650,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// send in chunks
int totalChunks;
int payloadChunkSize;
- if (canAddToBatch(msg) || !conf.isChunkingEnabled()) {
+ if (canAddToBatch(msg) || !isChunkingEnabled()) {
totalChunks = 1;
payloadChunkSize = getMaxMessageSize();
} else {
@@ -833,7 +836,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
MessageId messageId) throws IOException {
ByteBuf chunkPayload = compressedPayload;
MessageMetadata msgMetadata = msg.getMessageBuilder();
- if (totalChunks > 1 && TopicName.get(topic).isPersistent()) {
+ if (totalChunks > 1) {
chunkPayload = compressedPayload.slice(readStartIndex,
Math.min(chunkMaxSizeInBytes, chunkPayload.readableBytes() - readStartIndex));
// don't retain last chunk payload and builder as it will be not needed for next chunk-iteration and it will
@@ -3082,7 +3085,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
* Check if final message size for non-batch and non-chunked messages is larger than max message size.
*/
private PulsarClientException.InvalidMessageException getMessageSizeError(OpSendMsg op) {
- if (op.msg != null && !conf.isChunkingEnabled()) {
+ if (op.msg != null && !isChunkingEnabled()) {
int messageSize = op.getMessageHeaderAndPayloadSize();
if (messageSize > getMaxMessageSize()) {
return new PulsarClientException.InvalidMessageException(
@@ -3098,6 +3101,17 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
return getConnectionHandler().getMaxMessageSize();
}
+ /**
+ * Whether this producer actually chunks oversized messages. Chunking is only performed on persistent topics,
+ * so on a non-persistent topic the configuration flag is inert: the message is sent as a single frame and
+ * has to pass the same size checks as when chunking is disabled. Every decision that depends on chunking
+ * must go through this method rather than {@code conf.isChunkingEnabled()}, so that the size checks and the
+ * chunk computation cannot disagree.
+ */
+ private boolean isChunkingEnabled() {
+ return conf.isChunkingEnabled() && persistentTopic;
+ }
+
public long getDelayInMillis() {
OpSendMsg firstMsg = pendingMessages.peek();
if (firstMsg != null) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
index cf8cfb063287..c9e8c7c2d16a 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
@@ -977,4 +977,149 @@ public class BatchMessageContainerImplTest {
}
}
+ /**
+ * Multi-message batch over the maximum message size, with compression and encryption off (the default): the
+ * batch buffer is still the one the container owns when the size check fails, so {@code discard()} is its
+ * only owner and the oversized branch must not release it as well. A second release would drop a live
+ * buffer to zero and return it to the pool while it is still referenced.
+ */
+ @Test
+ public void testOversizedBatchReleasesItsPayloadExactlyOnce() throws Exception {
+ List<ByteBuf> allocated = new ArrayList<>();
+ ByteBufAllocator recordingAllocator = mock(ByteBufAllocator.class);
+ doAnswer(invocation -> {
+ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(invocation.getArgument(0));
+ allocated.add(buffer);
+ return buffer;
+ }).when(recordingAllocator).buffer(anyInt());
+
+ ProducerImpl<?> producer = createTestProducer();
+ // encryption disabled: the real implementation returns its input unchanged
+ doAnswer(invocation -> invocation.getArgument(1)).when(producer).encryptMessage(any(), any());
+ ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+ when(connectionHandler.getMaxMessageSize()).thenReturn(1);
+ when(producer.getConnectionHandler()).thenReturn(connectionHandler);
+
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(recordingAllocator);
+ container.setProducer(producer);
+ container.add(newMessage(1L, null, null), null);
+ container.add(newMessage(2L, null, null), null);
+
+ assertEquals(allocated.size(), 1, "expected exactly one batch payload allocation");
+ ByteBuf payload = allocated.get(0);
+ // Stand in for anything else still holding the payload, so that an extra release is observable rather
+ // than being swallowed by ReferenceCountUtil.safeRelease.
+ payload.retain();
+
+ try {
+ assertNull(container.createOpSendMsg(), "an oversized batch must not produce an op to send");
+
+ assertEquals(payload.refCnt(), 1, "the oversized batch payload was released more than once");
+ } finally {
+ ReferenceCountUtil.safeRelease(payload);
+ }
+ }
+
+ /**
+ * Multi-message batch over the maximum message size, with encryption on: a successful encryption releases
+ * its input and returns a different buffer, so ownership has left the container by the time the size check
+ * fails. The oversized branch must release the encrypted output exactly once, and {@code discard()} must not
+ * touch the pre-encryption buffer, which encryption already released.
+ */
+ @Test
+ public void testOversizedEncryptedBatchReleasesItsBuffersExactlyOnce() throws Exception {
+ List<ByteBuf> allocated = new ArrayList<>();
+ ByteBufAllocator recordingAllocator = mock(ByteBufAllocator.class);
+ doAnswer(invocation -> {
+ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(invocation.getArgument(0));
+ allocated.add(buffer);
+ return buffer;
+ }).when(recordingAllocator).buffer(anyInt());
+
+ ProducerImpl<?> producer = createTestProducer();
+ List<ByteBuf> encryptedBuffers = new ArrayList<>();
+ // Mirror a successful encryption: release the input and return a different buffer.
+ doAnswer(invocation -> {
+ ByteBuf input = invocation.getArgument(1);
+ ByteBuf encrypted = ByteBufAllocator.DEFAULT.buffer(input.readableBytes());
+ encrypted.writeBytes(input, input.readerIndex(), input.readableBytes());
+ input.release();
+ encryptedBuffers.add(encrypted);
+ return encrypted;
+ }).when(producer).encryptMessage(any(), any());
+ ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+ when(connectionHandler.getMaxMessageSize()).thenReturn(1);
+ when(producer.getConnectionHandler()).thenReturn(connectionHandler);
+
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(recordingAllocator);
+ container.setProducer(producer);
+ container.add(newMessage(1L, null, null), null);
+ container.add(newMessage(2L, null, null), null);
+
+ assertEquals(allocated.size(), 1, "expected exactly one batch payload allocation");
+ ByteBuf batchPayload = allocated.get(0);
+ // Stand in for anything else still holding the pre-encryption buffer, so that releasing it a second
+ // time is observable rather than being swallowed by ReferenceCountUtil.safeRelease.
+ batchPayload.retain();
+
+ try {
+ assertNull(container.createOpSendMsg(), "an oversized batch must not produce an op to send");
+
+ assertEquals(encryptedBuffers.size(), 1, "expected exactly one encryption");
+ assertEquals(batchPayload.refCnt(), 1,
+ "the pre-encryption batch payload was released again after encryption had already released it");
+ assertEquals(encryptedBuffers.get(0).refCnt(), 0, "the encrypted payload was not released");
+ } finally {
+ ReferenceCountUtil.safeRelease(batchPayload);
+ }
+ }
+
+ /**
+ * Single-message batch over the maximum message size, with compression and encryption off: the payload is
+ * handed to the SEND command before the size check runs, and {@code cmd.release()} releases it through the
+ * {@link ByteBufPair}. That is the only release the batch buffer may get; {@code discard()} afterwards must
+ * not release the same buffer again.
+ */
+ @Test
+ public void testOversizedSingleMessageBatchReleasesItsPayloadExactlyOnce() throws Exception {
+ List<ByteBuf> allocated = new ArrayList<>();
+ ByteBufAllocator recordingAllocator = mock(ByteBufAllocator.class);
+ doAnswer(invocation -> {
+ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(invocation.getArgument(0));
+ allocated.add(buffer);
+ return buffer;
+ }).when(recordingAllocator).buffer(anyInt());
+
+ ProducerImpl<?> producer = createTestProducer();
+ // encryption disabled: the real implementation returns its input unchanged
+ doAnswer(invocation -> invocation.getArgument(1)).when(producer).encryptMessage(any(), any());
+ // the single-message branch uses the (producerId, sequenceId, numMessages, messageId, ...) overload
+ doAnswer(invocation -> {
+ MessageMetadata metadata = invocation.getArgument(4);
+ ByteBuf payload = invocation.getArgument(5);
+ return Commands.newSend(0L, metadata.hasSequenceId() ? metadata.getSequenceId() : 0L, 1,
+ Commands.ChecksumType.Crc32c, metadata, payload);
+ }).when(producer).sendMessage(anyLong(), anyLong(), anyInt(), any(), any(), any());
+ ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+ when(connectionHandler.getMaxMessageSize()).thenReturn(1);
+ when(producer.getConnectionHandler()).thenReturn(connectionHandler);
+
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(recordingAllocator);
+ container.setProducer(producer);
+ container.add(newMessage(1L, null, null), null);
+
+ assertEquals(allocated.size(), 1, "expected exactly one batch payload allocation");
+ ByteBuf payload = allocated.get(0);
+ // Stand in for anything else still holding the payload, so that an extra release is observable rather
+ // than being swallowed by ReferenceCountUtil.safeRelease.
+ payload.retain();
+
+ try {
+ assertNull(container.createOpSendMsg(), "an oversized batch must not produce an op to send");
+
+ assertEquals(payload.refCnt(), 1, "the oversized batch payload was released more than once");
+ } finally {
+ ReferenceCountUtil.safeRelease(payload);
+ }
+ }
}