feat(java): support GZIP message body compression on producer side

[itxaiohanglover]

Which Issue(s) This PR Fixes

Fixes #1288

Brief Description

This PR adds opt-in producer-side message body compression for the Java client:

• New API ProducerBuilder#setCompressBodyThresholdBytes(int): once set, a message whose body size reaches the threshold is compressed with GZIP before sending. Compression is disabled by default (Integer.MAX_VALUE), so existing behavior is unchanged.
• PublishingMessageImpl now carries the transport body and its Encoding (IDENTITY / GZIP), and sets SystemProperties#bodyEncoding accordingly, instead of always hardcoding IDENTITY.
• The max body size check is applied to the transported (possibly compressed) body, matching the intent documented on PublishingSettings#maxBodySizeBytes ("it would be compressed for convenience of transport"), which was previously not implemented.

No change is required on the consuming side: MessageViewImpl already decompresses the body transparently according to the body encoding (magic-code based, GZIP supported).

Regarding zstd (mentioned in #1288): the protocol-level apache.rocketmq.v2.Encoding enum currently only defines IDENTITY and GZIP, so zstd support requires a protocol change in rocketmq-apis first. This PR takes GZIP as the first step; zstd can be a follow-up once the protocol supports it. (The compression utilities for ZSTD/ZLIB/LZ4 already exist in Utilities for the consuming side.)

How Did You Test This Change?

• Added PublishingMessageImplTest:
  • body is not compressed by default;
  • body reaching the threshold is compressed, bodyEncoding is GZIP, and Utilities#decompressBytes round-trips to the original body;
  • body below the threshold stays IDENTITY.
• Added builder validation tests in ProducerBuilderImplTest (negative threshold rejected).
• Ran the related unit tests locally: new tests 15/15 passed. (ProducerImplTest/TransactionImplTest Mockito errors on my machine are a pre-existing local JDK 21 incompatibility, reproduced on a clean master checkout as well.)

[RockteMQ-AI]

Review by github-manager-bot

Summary

Adds opt-in producer-side GZIP compression for message bodies in the Java client. When setCompressBodyThresholdBytes(int) is configured, messages whose body size reaches the threshold are compressed with GZIP before sending. The consumer already decompresses transparently via the Encoding magic-code, so no consumer-side changes are needed. Clean, backward-compatible design (disabled by default via Integer.MAX_VALUE).

Findings

• [Warning] ProducerBuilderImpl.java:104 — Encapsulation violation. producer.publishingSettings.compressBodyThresholdBytes = compressBodyThresholdBytes directly accesses a field on the internal PublishingSettings object from outside the class. This breaks the pattern used by other settings (which are set via constructor or sync()). Consider either:

  1. Adding a constructor parameter to PublishingSettings for the compression threshold, or
  2. Adding a package-private setter method on PublishingSettings instead of direct field access.

  This would also make the field easier to test in isolation and more resilient to future refactoring.

• [Info] PublishingMessageImpl.java:60-68 — Compression always wins over original. When body.length >= threshold, the compressed bytes are always used — even if GZIP makes the body larger (e.g., already-compressed or encrypted payloads). This is a reasonable simplification, but you might want to add a fallback:

  if (compressed.length < body.length) {
      transportBody = compressed;
      encoding = Encoding.GZIP;
  } else {
      transportBody = body;
      encoding = Encoding.IDENTITY;
  }

  This avoids wasting CPU on compression that increases size. If you intentionally skip this for simplicity, a brief comment in the code explaining the tradeoff would help future readers.

• [Info] PublishingMessageImpl.java:70-73 — Max body size check uses compressed size. The check transportBody.length > publishingSettings.getMaxBodySizeBytes() now validates the compressed body. This matches the documented intent on PublishingSettings#maxBodySizeBytes ("it would be compressed for convenience of transport"), which is good. Just worth noting that a message whose original body is under the limit could still be rejected if compression inflates it past maxBodySizeBytes.

• [Info] PublishingSettings.java — sync() does not touch compressBodyThresholdBytes. This is correct (it's a client-only setting, not pushed by the server), but worth adding a brief inline comment like // Client-only: not synced from server to prevent future confusion.

Test Coverage

Good coverage of the happy path. Consider adding:

1. A test where GZIP produces a larger body (e.g., random bytes) — verify the behavior is deterministic and documented.
2. A test where the compressed body exceeds maxBodySizeBytes — verify the message is rejected.
3. A boundary test: body.length == threshold (should compress per the >= check).

Cross-repo Note

The protocol Encoding enum (IDENTITY / GZIP) is defined in apache/rocketmq-apis. This PR correctly uses the existing enum values. No broker-side changes are needed since decompression happens on the consumer side. When zstd support is added later, it will require a protocol change in apache/rocketmq-apis first.

---

Automated review by github-manager-bot

[itxaiohanglover]

Thanks for the thorough review — all four points adopted in 621b7af:

1. Encapsulation: PublishingSettings#setCompressBodyThresholdBytes is now package-private, called only from ProducerBuilderImpl in the same package (tests go through a same-package helper).
2. Compression fallback: GZIP output is now used only when it is strictly smaller than the original body; otherwise the message falls back to IDENTITY, so already-compressed/encrypted payloads are never inflated.
3. sync() clarification: added an inline comment marking the threshold as a client-only setting that is never synced from the server.
4. Tests: added the three suggested cases — exact-threshold boundary (== threshold compresses), incompressible random body falls back to IDENTITY deterministically, and an oversized incompressible body is rejected against maxBodySizeBytes. Also added a case documenting that a >4 MiB compressible body is accepted when its compressed form fits, matching the documented transport-size semantics.

All 19 tests pass locally (PublishingMessageImplTest 7/7, ProducerBuilderImplTest 12/12).

[RockteMQ-AI]

Review by github-manager-bot (Follow-up)

New Commit: feat(java): address review — package-private setter, compression fallback, more tests

All previous review findings have been addressed:

1. Encapsulation ✅ — setCompressBodyThresholdBytes is now package-private. A dedicated PublishingSettingsTestHelper in the same package provides test-only access — clean pattern.

2. Compression fallback ✅ — PublishingMessageImpl now compares compressed vs. original size and falls back to IDENTITY encoding when GZIP inflates the body. This avoids wasting bandwidth on already-compressed or encrypted payloads.

3. Client-only comment ✅ — Added inline comment on setCompressBodyThresholdBytes clarifying it is never synced from the server.

4. Test coverage ✅ — Four new tests:

   • testBodyCompressedAtExactThreshold — boundary test at exact threshold (good edge case)
   • testBodyNotCompressedWhenGzipInflates — verifies fallback with incompressible random bytes
   • testIncompressibleBodyExceedingMaxSizeRejected — large incompressible body correctly rejected
   • testCompressibleBodyOverMaxSizeAcceptedWhenCompressedFits — large compressible body accepted when compressed size fits

Verdict

LGTM. The follow-up commit is well-structured, each change maps directly to a review finding, and the new tests cover the key edge cases. Ready for committer review.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Review by github-manager-bot

Summary

This PR adds opt-in producer-side GZIP message body compression for the Java client. When a message body exceeds the configured threshold, it is compressed before sending. Compression is disabled by default (Integer.MAX_VALUE), ensuring backward compatibility.

Findings

• [Info] ProducerBuilder.java:85 — The new API setCompressBodyThresholdBytes(int) is clean and follows the existing builder pattern. Javadoc is clear.

• [Info] ProducerBuilderImpl.java:89 — Validation compressBodyThresholdBytes > 0 is correct. Consider also documenting the behavior when threshold is very small (e.g., 1 byte) — every message will be compressed, which may hurt CPU for negligible size reduction.

• [Warning] PublishingMessageImpl.java:60-70 — The compression happens in the constructor, which is called per-message on the producer hot path. GZIP compression for large messages can be CPU-intensive. Consider:

  1. Adding a note in the Javadoc about the CPU/memory trade-off.
  2. Potentially using a shared GZIPOutputStream or buffered approach to reduce per-message allocation overhead.
  3. Documenting recommended threshold values (e.g., 1KB, 4KB) in the API doc.

• [Info] PublishingMessageImpl.java:49 — The BODY_ENCODING message property is a good approach for transparent consumer decompression. Ensure this property key is documented in the protocol spec or client compatibility matrix so other language clients can implement the same convention.

• [Info] PublishingSettings.java:42 — Thread safety with @GuardedBy("this") is consistent with the existing pattern. Good.

• [Info] Tests — Good coverage: builder validation, settings behavior, message compression/decompression roundtrip. The test helper PublishingSettingsTestHelper is a nice touch for keeping tests clean.

Suggestions

1. Consider adding a compression level parameter (1-9) in a future iteration, allowing users to tune the CPU vs. compression ratio trade-off.
2. The Encoding enum is package-private in message package. If other modules need to check encoding (e.g., interceptor chains), consider making it public or providing an accessor.

Cross-repo Note

The BODY_ENCODING property convention should be coordinated with the Proxy module in apache/rocketmq to ensure the Proxy does not strip or alter this message property during forwarding.

---

Automated review by github-manager-bot