fix: allocate rebatch buffer with uncompressed size (#26290)

[waterWang]

Motivation

RawBatchConverter.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

Move metadata.getUncompressedSize() before the buffer allocation and use it as the initial capacity instead of payload.capacity().

Affected versions

• 4.0.x (LTS) and 4.1.x: the bug manifests as a deterministic compaction failure for compressed batches.
• 4.2.0+ / master: the LightProto 0.6.x upgrade ([improve][build] Upgrade LightProto to 0.6.1 #25332) masks the bug by making the generated SingleMessageMetadata.writeTo() call ensureWritable() before the zero-copy getBytes path. This fix is still a defensive improvement for these versions.

Verification

The reproducer from the issue (a topic with batched ZSTD-compressed messages and automatic compaction) can be used to verify the fix.

Fixes #26290

[hozumi]

Thanks for picking this up! Claude Fable 5 review below:

Unfortunately uncompressedSize is not a sufficient upper bound: a compacted-out message becomes an ~8-byte placeholder (4B size prefix + ~4B compactedOut metadata), which is larger than a keyless tiny message (7B for a 1-byte payload; keyless messages are compacted out under the default topicCompactionRetainNullKey=false). The rebatched output can then exceed uncompressedSize, and a later kept keyed message's zero-copy key write overruns the buffer — same exception, no compression involved. On master the generated writeTo() masks this via ensureWritable, but on branch-4.0 / branch-4.1 — where this fix matters most — it regresses a case that currently works:

allocation	issue 26290 reproducer (ZSTD)	reproducer below (mixed, no compression)
payload.capacity() (current)	FAIL (dstIndex: 4347)	success
uncompressedSize (this PR)	success	FAIL (dstIndex: 4013)
uncompressedSize + 8 * metadata.getNumMessagesInBatch()	success	success

Verified with the reproducer below — it applies this PR's allocation change to the 4.0.13 sources and loads the class via the broker classpath:

docker run -d --name pulsar-pr-repro -u 0 apachepulsar/pulsar:4.0.13 bash -c '
  curl -sLo /tmp/RawBatchConverter.java https://raw.githubusercontent.com/apache/pulsar/v4.0.13/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
  sed -i "s/buffer(payload.capacity())/buffer(metadata.getUncompressedSize())/" /tmp/RawBatchConverter.java
  javac -nowarn -cp "/pulsar/lib/*" -d /tmp/patch /tmp/RawBatchConverter.java
  PULSAR_CLASSPATH=/tmp/patch exec bin/pulsar standalone -nss -nfw'
docker exec pulsar-pr-repro bash -c 'until bin/pulsar-admin brokers healthcheck >/dev/null 2>&1; do sleep 3; done'

docker exec -i pulsar-pr-repro python3 - <<'EOF'
import pulsar
client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer(
    "persistent://public/default/compaction-mixed-repro",
    batching_enabled=True,
    batching_max_messages=1000,
    batching_max_publish_delay_ms=1000,
    block_if_queue_full=True,
)
# one batch: 200 keyless 1-byte payloads (compacted out -> 8B placeholders > 7B originals),
# then 800 keyed messages (kept -> zero-copy key writes)
for i in range(200):
    producer.send_async(b"x", callback=lambda res, mid: None)
for i in range(800):
    producer.send_async(b"payload-%d" % i, callback=lambda res, mid: None,
                        partition_key="%08d/%s" % (i, "k" * 120))
producer.flush()
client.close()
print("produced")
EOF

docker exec pulsar-pr-repro bin/pulsar-admin topics compact persistent://public/default/compaction-mixed-repro
sleep 5
docker exec pulsar-pr-repro bin/pulsar-admin topics compaction-status persistent://public/default/compaction-mixed-repro
# with this PR's allocation: "Error compacting: java.lang.IndexOutOfBoundsException: dstIndex: 4013"
# drop the curl/sed/javac lines (= stock 4.0.13): "Compaction was a success"