[improve][broker] Bound decompressed message size before allocation; release ByteBuf on codec exceptions
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Motivation
When a message is compressed, the decompressed size is taken directly from the uncompressedSize field of MessageMetadata and used to allocate a ByteBuf before decompression. This field is client-controlled, the broker's produce path does not validate it before persisting the message, and most decompression call-sites do not validate it before allocating. A single malformed message can therefore trigger an allocation of up to ~2 GiB (Integer.MAX_VALUE) on any component that later decompresses the entry — even though the compressed payload itself is tiny and passes the frame-size limit.
This is filed as a robustness / defense-in-depth improvement. (Per the documented security model, denial of service by a perimeter-trusted client is outside Pulsar's threat model, so this is not a security report.)
### Solution
1. Bound uncompressedSize at the produce entry point (preferred — prevents the malformed entry from being persisted): in ServerCnx.handleSend / PersistentTopic.publishMessage, parse the metadata once and reject the send when uncompressedSize > maxMessageSize.
2. Defense-in-depth at every codec.decode(..., getUncompressedSize()) call-site: in MessageParser.uncompressPayloadIfNeeded (add an uncompressedSize > maxMessageSize branch alongside the existing payloadSize check at :160), PersistentTopicsBase:3350, PulsarTopicCompactionService:150, AbstractTwoPhaseCompactor:479, RawBatchConverter:69/150. This also covers entries that bypass the produce path (cross-cluster replication from older brokers, direct BookKeeper tampering).
3. Release ByteBuf on codec exceptions: wrap the allocation + (de)compression in try/finally, releasing the buffer on failure, in CompressionCodecLZ4/Zstd/ZLib/Snappy.encode/decode.
### Alternatives
_No response_
### Anything else?
_No response_
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with ServerCnx.handleSend and PersistentTopic.publishMessage, then inspect the listed codec.decode call-sites, especially MessageParser.uncompressPayloadIfNeeded at :160. Review the allocation and exception paths in CompressionCodecLZ4, CompressionCodecZstd, CompressionCodecZLib, and CompressionCodecSnappy. Done means oversized uncompressed sizes are rejected before allocation or persistence, and allocated buffers are released when codec operations fail.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems, performance
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 45/100