[Bug] Topic compaction fails permanently with IndexOutOfBoundsException when rebatching compressed batch messages

### Search before reporting

- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Read release policy

- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

### User environment

Where we hit the issue (reproduction environments are described in the reproducing section below):

- Broker: 4.0.8, as a docker container on Rocky Linux 8 (x86_64)
- Broker Java: the JVM bundled in the broker image
- Client library: Java client 4.0.10 (the producer of the affected topic). The client version is not relevant — the bug is broker-side.

### Issue Description

> **Disclosure**: I hit and confirmed this issue in our production cluster myself. The investigation, the root-cause analysis, the minimal reproducer, the version matrix, the fix verification, and most of this write-up were done by an AI agent (Claude Fable 5 via Claude Code) working under my direction. I have reviewed this report and run the reproducer.

**What happened**: On a topic receiving keyed messages from a producer with **batching enabled + ZSTD compression**, topic compaction fails with

```
java.lang.IndexOutOfBoundsException: dstIndex: 4347
```

thrown from `RawBatchConverter.rebatchMessage` during phase two. The failure is deterministic for a given ledger entry (same `dstIndex` on every attempt), so compaction on the topic **never succeeds again**.

**Expected**: compaction succeeds.

**Impact in production**: this is much worse than a single failed compaction. With automatic compaction enabled (`compactionThreshold` set on the namespace), the broker retries every `brokerServiceCompactionMonitorIntervalInSeconds` (60s default), and **each attempt re-reads the whole topic in phase one before failing in phase two**. On our production topic (~14.6M messages / ~20k batch entries / 165 MB), this produced a constant ~20M msg/min of internal dispatch (`msgOutCounter` reached 59.6 billion, `bytesOutCounter` 677 GB over 3 days) until we disabled automatic compaction with `set-compaction-threshold --threshold 0`.

**Root cause analysis**: `RawBatchConverter.rebatchMessage` allocates the output buffer sized by the **compressed** payload, then re-serializes the **uncompressed** single messages into it:

```java
// RawBatchConverter.rebatchMessage — payload is still compressed here
ByteBuf batchBuffer = PulsarByteBufAllocator.DEFAULT.buffer(payload.capacity());
...
ByteBuf uncompressedPayload = codec.decode(payload, uncompressedSize);
```

Most writes into `batchBuffer` go through auto-expanding methods (`writeInt`, `writeByte`, `writeBytes`), so an undersized initial capacity is usually harmless. However, the partition key (and ordering key) write in the lightproto-generated `SingleMessageMetadata.writeTo` uses the zero-copy path when the field was parsed from a buffer:

```java
// generated SingleMessageMetadata.writeTo
_parsedBuffer.getBytes(_partitionKeyBufferIdx, _b, _partitionKeyBufferLen);
```

Netty's `ByteBuf.getBytes(int index, ByteBuf dst, int length)` does **not** expand the destination — it throws `IndexOutOfBoundsException: dstIndex: ` when `writerIndex + length > dst.capacity()`.

So whenever a batch compresses well (compressed size ≪ uncompressed size) and a capacity boundary of `batchBuffer` happens to land inside a partition-key write, rebatching fails deterministically. With many messages per batch and keys making up a large fraction of the bytes, hitting such a boundary is nearly certain (the reproducer below fails on the first attempt). This should affect any compression codec, not just ZSTD, since the only requirement is `compressed capacity < uncompressed rebatched size`.

4.2.0+ and master are unaffected as a side effect of the LightProto 0.6.x upgrade ([#25332](https://github.com/apache/pulsar/pull/25332)): the regenerated `writeTo()` calls `_b.ensureWritable(_serializedSize)` up front (verified in the generated sources), so the undersized allocation is silently grown. Release lines still on lightproto-maven-plugin 0.4 — **4.0.x (LTS) and 4.1.x** — are affected, matching the empirical version matrix in the reproducing section.

### Error messages

```text
$ docker exec pulsar-compaction-repro bin/pulsar-admin topics compaction-status persistent://public/default/compaction-zstd-repro
Error in compaction
null

Reason: Error compacting: java.lang.IndexOutOfBoundsException: dstIndex: 4347

Broker log:

2026-08-08T03:11:07,121+0000 [broker-client-shared-internal-executor-21-1] WARN org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://public/default/compaction-zstd-repro] Compaction failure. {}
java.util.concurrent.CompletionException: java.lang.IndexOutOfBoundsException: dstIndex: 4347
...
Caused by: java.lang.IndexOutOfBoundsException: dstIndex: 4347
at io.netty.buffer.UnsafeByteBufUtil.getBytes(UnsafeByteBufUtil.java:486)
at io.netty.buffer.PooledUnsafeDirectByteBuf.getBytes(PooledUnsafeDirectByteBuf.java:124)
at io.netty.buffer.AbstractByteBuf.getBytes(AbstractByteBuf.java:502)
at org.apache.pulsar.common.api.proto.SingleMessageMetadata.writeTo(SingleMessageMetadata.java:329)
at org.apache.pulsar.common.protocol.Commands.serializeSingleMessageInBatchWithPayload(Commands.java:1870)
at org.apache.pulsar.client.impl.RawBatchConverter.rebatchMessage(RawBatchConverter.java:182)
at org.apache.pulsar.compaction.AbstractTwoPhaseCompactor.rebatchMessage(AbstractTwoPhaseCompactor.java:481)
at org.apache.pulsar.compaction.AbstractTwoPhaseCompactor.lambda$phaseTwoLoop$19(AbstractTwoPhaseCompactor.java:295)
```

### Reproducing the issue

Fully self-contained — only docker is required (the producer runs on the image's bundled Python client). Produce keyed messages with **batching enabled + ZSTD compression** where the batch content is highly compressible (repetitive keys, small payloads), then trigger compaction:

```bash
docker run -d --name pulsar-compaction-repro apachepulsar/pulsar:4.0.13 bin/pulsar standalone -nss -nfw
docker exec pulsar-compaction-repro bash -c 'until bin/pulsar-admin brokers healthcheck >/dev/null 2>&1; do sleep 3; done'

docker exec -i pulsar-compaction-repro python3 - <<'EOF'
import pulsar
client = pulsar.Client("pulsar://localhost:6650")
producer = client.create_producer(
"persistent://public/default/compaction-zstd-repro",
compression_type=pulsar.CompressionType.ZSTD,
batching_enabled=True,
batching_max_messages=1000,
batching_max_publish_delay_ms=1000,
block_if_queue_full=True,
)
for i in range(5000):
# payloads must be non-empty: empty payloads are tombstones and skip the
# partition-key write during rebatch
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 5000 messages")
EOF

docker exec pulsar-compaction-repro bin/pulsar-admin topics compact persistent://public/default/compaction-zstd-repro
sleep 5
docker exec pulsar-compaction-repro bin/pulsar-admin topics compaction-status persistent://public/default/compaction-zstd-repro
```

To test another broker version, change the image tag. Control experiment: producing the exact same messages **without compression** (drop the `compression_type` argument) and compacting succeeds, which isolates the trigger to batching + compression.

Results of the exact same reproducer against multiple broker versions:

| Broker | lightproto codegen | Result |
|---|---|---|
| 4.0.8 | 0.4 | **FAIL** — `IndexOutOfBoundsException: dstIndex: 4347` |
| 4.0.13 | 0.4 | **FAIL** — `IndexOutOfBoundsException: dstIndex: 4347` |
| 4.1.3 | 0.4 | **FAIL** — `IndexOutOfBoundsException: dstIndex: 4347` |
| 4.2.4 | 0.6.2 | success |
| 5.0.0-M1 | 0.7.3 | success |

(the identical `dstIndex` across versions shows the failure is fully deterministic for a given input)

### Additional information

### Suggested fix

Two possible approaches, in root-cause order:

**A. Backport the LightProto upgrade ([#25332](https://github.com/apache/pulsar/pull/25332)) to `branch-4.0` / `branch-4.1`.** This is what makes 4.2.0+ safe: the regenerated `writeTo()` ensures capacity up front, fixing the fragile contract at its source. Since the generated sources are not committed, the diff is essentially a plugin version bump — but it regenerates every wire-protocol serialization class on a maintenance branch.

**B. If regenerating the protocol classes on maintenance branches is not acceptable: size the buffer from the uncompressed size in `RawBatchConverter.rebatchMessage`** (arguably also worth applying on `master` as defense in depth):

```java
ByteBuf batchBuffer = PulsarByteBufAllocator.DEFAULT.buffer(
Math.max(payload.capacity(),
metadata.getUncompressedSize() + 8 * metadata.getNumMessagesInBatch()));
```

Kept messages are re-serialized identically, and filtered-out messages are replaced by a `compactedOut` placeholder (4-byte size prefix + ~4-byte metadata), which can exceed a minimal original single message by a few bytes — hence the small per-message headroom. Sizing from a computed upper bound matches the idiom of the similar serialization sites nearby (`RawBatchMessageContainerImpl#toByteBuf`, `RawMessageImpl#serialize`).

**Option B is verified against the reproducer**: compiling the 4.0.13 `RawBatchConverter.java` with only the allocation line changed as above, and prepending it to the broker classpath of the `apachepulsar/pulsar:4.0.13` image (`PULSAR_CLASSPATH=/patch/classes`), makes the exact same reproducer pass:

```
$ curl -s http://localhost:8080/admin/v2/persistent/public/default/compaction-zstd-repro/compaction
{"status":"SUCCESS","lastError":""}
```

with the JVM class-load log confirming the patched class was the one in use:

```
[13.169s][info][class,load] org.apache.pulsar.client.impl.RawBatchConverter source: file:/patch/classes/
```

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Start with org.apache.pulsar.client.impl.RawBatchConverter.rebatchMessage and the compaction stack shown in the broker trace, then run the self-contained Docker reproducer against an affected 4.0.x or 4.1.x image. Compare the compressed and uncompressed cases and verify that compaction completes successfully without the IndexOutOfBoundsException on the affected release line.

Written by the indexing model from the issue text.