From 24342de5b21ef248603b920398f2b61bf1badaa6 Mon Sep 17 00:00:00 2001
From: water <672684719@qq.com>
Date: Sun, 9 Aug 2026 07:06:52 +0800
Subject: [PATCH] fix: allocate rebatch buffer with uncompressed size

rebatchMessage allocates the output batch buffer using the compressed
payload capacity, but then re-serializes the *uncompressed* single
messages into it. When the compressed payload is much smaller than the
uncompressed rebatched size, the buffer is undersized and
SingleMessageMetadata.writeTo's zero-copy getBytes path (from
_parsedBuffer) throws IndexOutOfBoundsException because Netty's
getBytes does not auto-expand the destination.

Fix: size the buffer from metadata.getUncompressedSize() so rebatching
never writes past the initial capacity. Affects compressed batches on
4.0.x / 4.1.x (LTS); 4.2.0+ masks the issue via the LightProto 0.6.x
ensureWritable upgrade, but the allocation is still incorrect there.

Fixes #26290
---
 .../java/org/apache/pulsar/client/impl/RawBatchConverter.java | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
index 64ddea3ec6ab1..8ef76f64b0842 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
@@ -142,12 +142,12 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
         } else {
             Commands.skipMessageMetadata(payload);
         }
-        ByteBuf batchBuffer = PulsarByteBufAllocator.DEFAULT.buffer(payload.capacity());
+        int uncompressedSize = metadata.getUncompressedSize();
+        ByteBuf batchBuffer = PulsarByteBufAllocator.DEFAULT.buffer(uncompressedSize);
 
         CompressionType compressionType = metadata.getCompression();
         CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(compressionType);
 
-        int uncompressedSize = metadata.getUncompressedSize();
         ByteBuf uncompressedPayload = codec.decode(payload, uncompressedSize);
         try {
             int batchSize = metadata.getNumMessagesInBatch();