Search before reporting
Read release policy
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:
// 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:
// 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: <writerIndex> 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): 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
$ 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:
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-reproTo 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) 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):
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?
Search before reporting
Read release policy
User environment
Where we hit the issue (reproduction environments are described in the reproducing section below):
Issue Description
What happened: On a topic receiving keyed messages from a producer with batching enabled + ZSTD compression, topic compaction fails with
thrown from
RawBatchConverter.rebatchMessageduring phase two. The failure is deterministic for a given ledger entry (samedstIndexon 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 (
compactionThresholdset on the namespace), the broker retries everybrokerServiceCompactionMonitorIntervalInSeconds(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 (msgOutCounterreached 59.6 billion,bytesOutCounter677 GB over 3 days) until we disabled automatic compaction withset-compaction-threshold --threshold 0.Root cause analysis:
RawBatchConverter.rebatchMessageallocates the output buffer sized by the compressed payload, then re-serializes the uncompressed single messages into it:Most writes into
batchBuffergo 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-generatedSingleMessageMetadata.writeTouses the zero-copy path when the field was parsed from a buffer:Netty's
ByteBuf.getBytes(int index, ByteBuf dst, int length)does not expand the destination — it throwsIndexOutOfBoundsException: dstIndex: <writerIndex>whenwriterIndex + length > dst.capacity().So whenever a batch compresses well (compressed size ≪ uncompressed size) and a capacity boundary of
batchBufferhappens 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 iscompressed capacity < uncompressed rebatched size.4.2.0+ and master are unaffected as a side effect of the LightProto 0.6.x upgrade (#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
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:
To test another broker version, change the image tag. Control experiment: producing the exact same messages without compression (drop the
compression_typeargument) and compacting succeeds, which isolates the trigger to batching + compression.Results of the exact same reproducer against multiple broker versions:
IndexOutOfBoundsException: dstIndex: 4347IndexOutOfBoundsException: dstIndex: 4347IndexOutOfBoundsException: dstIndex: 4347(the identical
dstIndexacross 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) to
branch-4.0/branch-4.1. This is what makes 4.2.0+ safe: the regeneratedwriteTo()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 onmasteras defense in depth):Kept messages are re-serialized identically, and filtered-out messages are replaced by a
compactedOutplaceholder (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.javawith only the allocation line changed as above, and prepending it to the broker classpath of theapachepulsar/pulsar:4.0.13image (PULSAR_CLASSPATH=/patch/classes), makes the exact same reproducer pass:with the JVM class-load log confirming the patched class was the one in use:
Are you willing to submit a PR?