leizhiyuan opened a new pull request, #1429:
URL: https://github.com/apache/rocketmq-clients/pull/1429
### Which Issue(s) This PR Fixes
Fixes #1428
### Brief Description
**Scenario.** During a migration to 5.x, producers often still use the
Remoting SDK while new consumers use the gRPC clients. The Remoting producer
compresses bodies over 4 KiB with ZLIB by default, and the proxy delivers them
with `body_encoding = GZIP`. The Rust client ignores `body_encoding` and
`body_digest`, so `MessageView::body()` returns the compressed bytes (a Java
consumer of the same group gets the original text) and a body that fails its
digest is delivered as valid.
**Change.** Align `MessageView::from_pb_message` and the push consumer with
Java (`MessageViewImpl#fromProtobuf`, `Utilities#decompressBytes`,
`StandardConsumeService`, `FifoConsumeService#getNextValidMessage`):
- **Digest.** Verify `body_digest` (CRC32 / MD5 / SHA1) on the body as
received, before decoding. The proxy computes it on the stored, still
compressed body. The checksum strings are built the same way as Java: uppercase
hex, CRC32 without zero padding (`Long.toHexString`).
- **Decoding.** When `body_encoding == GZIP`, decompress by magic number:
`1f 8b` → GZIP, `78` → ZLIB. The proxy labels every compressed body `GZIP`,
whatever the algorithm, so ZLIB has to be handled too.
- **Corrupted flag.** Add `MessageView::is_corrupted()`, true when the
digest does not match or the body can not be decompressed. A corrupted message
keeps the body as received. Same as Java `MessageView#isCorrupted`.
- **Push consumer.** A corrupted message is not passed to the listener. The
standard worker nacks it with the retry policy's next delay. The FIFO worker
forwards it to the dead letter queue. This covers both the sequential path and
the FIFO consume accelerator, since both go through
`consume_message_sequentially_static`.
- **Simple consumer.** Returns corrupted messages as before; the caller can
check `is_corrupted()`, as in Java.
#### Notes for reviewers
- **New dependencies.** `flate2 = "1"` (default `miniz_oxide` backend, pure
Rust), `crc32fast = "1"` (already a dependency of `flate2`), `md-5 = "0.10"`.
SHA-1 uses the existing `ring` dependency (`SHA1_FOR_LEGACY_USE_ONLY`), and hex
encoding the existing `hex` dependency. All three are MIT/Apache-2.0. I checked
that they build with Rust 1.74 (the MSRV used by CI) in a standalone project;
`master` itself no longer builds on 1.74 because of `tokio` 1.54, which is the
existing red MSRV check.
- **Not supported: LZ4 / ZSTD.** Java also recognizes LZ4 and ZSTD magic
numbers. The Remoting producer uses ZLIB unless `rocketmq.message.compressType`
is changed, so this PR handles GZIP and ZLIB and marks other formats as
corrupted (logged as "unsupported compression format"). Support can be added
later without changing the API.
- **Behavior change.** Consumers now see the decompressed body of
`GZIP`-encoded messages. A push consumer no longer passes corrupted messages to
the listener.
- **Conflicts.** `git merge-tree` is clean against #1381, #1404, #1407,
#1409, #1416, #1420, #1421, #1423, #1425 and #1427. It has textual conflicts in
`push_consumer.rs` with #1411 and #1414:
- #1411 changes the listener call in the standard worker to
`consume_message(..)`. Resolution: keep the corrupted check before it.
- #1414 rewrites the FIFO retry loop. Resolution: keep the corrupted check
before the loop and read `message.delivery_attempt()` there.
- Both also append tests at the end of the test module; keep both.
- I verified both on temporary merges: `cargo test` 99 passed (with #1411)
and 100 passed (with #1414). Overview in #1412.
### How Did You Test This Change?
**Unit tests.** All of them fail on `master` except `identity_body_is_kept`:
- `model::message::tests::body_decoding`
- `identity_body_is_kept`
- `gzip_body_is_decompressed`
- `zlib_body_marked_as_gzip_is_decompressed`
- `undecodable_body_is_corrupted`: a truncated gzip stream and an unknown
format; the body is kept as received.
- `digest_is_verified_on_the_received_body`: an MD5 over the compressed
body passes, an MD5 over the plain body is corrupted.
- `every_digest_type_is_verified`: CRC32, MD5, SHA1, matching and not
matching.
- `push_consumer::tests`
- `test_standard_consume_discards_corrupted_message`: the listener is not
called, `change_invisible_duration` once, no ack.
- `test_fifo_consume_forwards_corrupted_message_to_dlq`: the listener is
not called, `forward_to_deadletter_queue` once, no ack.
```
$ cargo test -- --test-threads=1
test result: ok. 97 passed; 0 failed; 1 ignored (lib; master: 89)
test result: ok. 3 passed; 0 failed (lite_integration_test)
test result: ok. 2 passed; 0 failed; 3 ignored (doc-tests)
$ cargo fmt --check # clean
$ RUSTDOCFLAGS='-D warnings' cargo doc --no-deps # no warnings
$ cargo build --examples --all-features # ok
```
`cargo clippy --all-targets` reports the same findings as `master`.
`--test-threads=1` avoids the existing flaky
`client::tests::client_get_session` (it uses `MockSession::new_context()`
without holding `MTX`), unrelated to this change.
**End-to-end.** Against a managed RocketMQ 5 server, with a normal topic and
group, received with `SimpleConsumer`:
| message | before | after | Java 5.2.2 |
|---|---|---|---|
| 5090-byte text from the Remoting producer (`rocketmq-client` 5.3.1,
compressed because > 4 KiB) | 472 bytes, `78 5e ..` (ZLIB) | the original
5090-byte text, not corrupted | the original text |
| 15-byte text from a gRPC producer with `body_encoding = GZIP` | the
35-byte gzip stream | the original 15-byte text, not corrupted | the original
text |
Both messages carry the proxy's MD5 digest, which now passes. The other
SimpleConsumer end-to-end cases (send/receive/ack, change invisible duration,
tag and SQL filters, idle long polling, delay message) behave as before.
---
Tracked in #1412 (overview of the related Rust issues / PRs and a suggested
merge order).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
