This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 9f77cf18 [ISSUE #1428] [Rust] Decompress received message body and
verify its digest (#1429)
9f77cf18 is described below
commit 9f77cf18a8c049698d97f3a1268626debdf3f549
Author: Lei Zhiyuan <[email protected]>
AuthorDate: Fri Oct 23 22:20:19 2026 +0800
[ISSUE #1428] [Rust] Decompress received message body and verify its digest
(#1429)
---
rust/Cargo.toml | 3 +
rust/src/model/message.rs | 242 +++++++++++++++++++++++++++++++++++++++++++++-
rust/src/push_consumer.rs | 164 +++++++++++++++++++++++++++++++
3 files changed, 408 insertions(+), 1 deletion(-)
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
index dc8ca8d7..14968994 100644
--- a/rust/Cargo.toml
+++ b/rust/Cargo.toml
@@ -62,6 +62,9 @@ mockall = "0.11.4"
mockall_double = "0.3.0"
siphasher = "0.3.10"
+flate2 = "1"
+crc32fast = "1"
+md-5 = "0.10"
ring = "0.16.20"
tokio-util = { version = "=0.7.10", features = ["rt"] }
# Pin indexmap to avoid hashbrown 0.17+ which requires Rust 2024 edition
diff --git a/rust/src/model/message.rs b/rust/src/model/message.rs
index 5c58c2f6..b7aaf8a4 100644
--- a/rust/src/model/message.rs
+++ b/rust/src/model/message.rs
@@ -18,6 +18,9 @@
//! Message data model of RocketMQ rust client.
use std::collections::HashMap;
+use std::io::Read;
+
+use tracing::error;
use crate::error::{ClientError, ErrorKind};
use crate::model::common::Endpoints;
@@ -431,6 +434,91 @@ impl MessageBuilder {
}
}
+/// Verify the body digest and decode the body according to its encoding.
+///
+/// Returns the body to expose and whether the message is corrupted. The
digest is computed by
+/// the server on the body as stored, so it is checked before decoding.
+///
+/// Reference Java: `MessageViewImpl#fromProtobuf`.
+fn decode_body(
+ body: Vec<u8>,
+ system_properties: &pb::SystemProperties,
+ topic: &str,
+ message_id: &str,
+) -> (Vec<u8>, bool) {
+ let mut corrupted = false;
+ if let Some(digest) = &system_properties.body_digest {
+ match pb::DigestType::from_i32(digest.r#type) {
+ Some(pb::DigestType::Crc32) => {
+ corrupted |= format!("{:X}", crc32fast::hash(&body)) !=
digest.checksum;
+ }
+ Some(pb::DigestType::Md5) => {
+ use md5::Digest;
+ corrupted |= hex::encode_upper(md5::Md5::digest(&body)) !=
digest.checksum;
+ }
+ Some(pb::DigestType::Sha1) => {
+ let sha1 =
ring::digest::digest(&ring::digest::SHA1_FOR_LEGACY_USE_ONLY, &body);
+ corrupted |= hex::encode_upper(sha1.as_ref()) !=
digest.checksum;
+ }
+ _ => error!(
+ "Unsupported message body digest algorithm, digest_type={},
topic={}, message_id={}",
+ digest.r#type, topic, message_id
+ ),
+ }
+ }
+ if corrupted {
+ error!(
+ "Message body digest mismatch, topic={}, message_id={}",
+ topic, message_id
+ );
+ }
+ match pb::Encoding::from_i32(system_properties.body_encoding) {
+ Some(pb::Encoding::Gzip) => match decompress(&body) {
+ Ok(decompressed) => (decompressed, corrupted),
+ Err(e) => {
+ error!(
+ "Failed to uncompress message body, topic={},
message_id={}, error={}",
+ topic, message_id, e
+ );
+ (body, true)
+ }
+ },
+ Some(pb::Encoding::Identity) => (body, corrupted),
+ _ => {
+ error!(
+ "Unsupported message encoding algorithm, topic={},
message_id={}, body_encoding={}",
+ topic, message_id, system_properties.body_encoding
+ );
+ (body, corrupted)
+ }
+ }
+}
+
+/// Decompress a body by its magic number.
+///
+/// The proxy marks every body that was stored compressed as `GZIP`, whatever
the algorithm. The
+/// remoting producer compresses with ZLIB by default. LZ4 and ZSTD are not
supported.
+///
+/// Reference Java: `Utilities#decompressBytes`.
+fn decompress(body: &[u8]) -> std::io::Result<Vec<u8>> {
+ let mut decompressed = Vec::new();
+ match body {
+ [0x1f, 0x8b, ..] => {
+ flate2::read::MultiGzDecoder::new(body).read_to_end(&mut
decompressed)?;
+ }
+ [0x78, ..] => {
+ flate2::read::ZlibDecoder::new(body).read_to_end(&mut
decompressed)?;
+ }
+ _ => {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "unsupported compression format",
+ ))
+ }
+ }
+ Ok(decompressed)
+}
+
/// [`AckMessageEntry`] is the data model for ack message.
pub trait AckMessageEntry {
fn topic(&self) -> String;
@@ -460,6 +548,7 @@ pub struct MessageView {
pub(crate) endpoints: Endpoints,
pub(crate) priority: Option<i32>,
pub(crate) lite_topic: Option<String>,
+ pub(crate) corrupted: bool,
}
impl AckMessageEntry for MessageView {
@@ -484,12 +573,18 @@ impl MessageView {
pub(crate) fn from_pb_message(message: pb::Message, endpoints: Endpoints)
-> Option<Self> {
let system_properties = message.system_properties?;
let topic = message.topic?;
+ let (body, corrupted) = decode_body(
+ message.body,
+ &system_properties,
+ &topic.name,
+ &system_properties.message_id,
+ );
Some(MessageView {
message_id: system_properties.message_id,
receipt_handle: system_properties.receipt_handle,
namespace: topic.resource_namespace,
topic: topic.name,
- body: message.body,
+ body,
tag: system_properties.tag,
keys: system_properties.keys,
properties: message.user_properties,
@@ -501,6 +596,7 @@ impl MessageView {
endpoints,
priority: system_properties.priority,
lite_topic: system_properties.lite_topic,
+ corrupted,
})
}
@@ -524,6 +620,13 @@ impl MessageView {
&self.body
}
+ /// Whether the message body failed its digest check or could not be
decompressed.
+ ///
+ /// A corrupted message keeps the body as received. Reference Java:
`MessageViewImpl#isCorrupted`.
+ pub fn is_corrupted(&self) -> bool {
+ self.corrupted
+ }
+
/// Get message tag
pub fn tag(&self) -> Option<&str> {
self.tag.as_deref()
@@ -587,6 +690,143 @@ impl MessageView {
mod tests {
use super::*;
+ mod body_decoding {
+ use std::io::Write;
+
+ use super::*;
+
+ const PLAIN: &[u8] = b"hello compressed body hello compressed body
hello compressed body";
+
+ fn gzip(data: &[u8]) -> Vec<u8> {
+ let mut encoder =
+ flate2::write::GzEncoder::new(Vec::new(),
flate2::Compression::default());
+ encoder.write_all(data).unwrap();
+ encoder.finish().unwrap()
+ }
+
+ fn zlib(data: &[u8]) -> Vec<u8> {
+ let mut encoder =
+ flate2::write::ZlibEncoder::new(Vec::new(),
flate2::Compression::default());
+ encoder.write_all(data).unwrap();
+ encoder.finish().unwrap()
+ }
+
+ fn md5_hex(data: &[u8]) -> String {
+ use md5::Digest;
+ hex::encode_upper(md5::Md5::digest(data))
+ }
+
+ fn view(body: Vec<u8>, encoding: pb::Encoding, digest:
Option<pb::Digest>) -> MessageView {
+ MessageView::from_pb_message(
+ pb::Message {
+ topic: Some(pb::Resource {
+ name: "topic".to_string(),
+ resource_namespace: "".to_string(),
+ }),
+ user_properties: Default::default(),
+ system_properties: Some(pb::SystemProperties {
+ body_encoding: encoding as i32,
+ body_digest: digest,
+ ..Default::default()
+ }),
+ body,
+ },
+ Endpoints::from_url("localhost:8081").unwrap(),
+ )
+ .unwrap()
+ }
+
+ fn digest(r#type: pb::DigestType, checksum: String) ->
Option<pb::Digest> {
+ Some(pb::Digest {
+ r#type: r#type as i32,
+ checksum,
+ })
+ }
+
+ #[test]
+ fn identity_body_is_kept() {
+ let message = view(PLAIN.to_vec(), pb::Encoding::Identity, None);
+ assert_eq!(message.body(), PLAIN);
+ assert!(!message.is_corrupted());
+ }
+
+ #[test]
+ fn gzip_body_is_decompressed() {
+ let message = view(gzip(PLAIN), pb::Encoding::Gzip, None);
+ assert_eq!(message.body(), PLAIN);
+ assert!(!message.is_corrupted());
+ }
+
+ #[test]
+ fn zlib_body_marked_as_gzip_is_decompressed() {
+ // The proxy marks every body stored with COMPRESSED_FLAG as GZIP,
but the remoting
+ // producer compresses with ZLIB by default.
+ let message = view(zlib(PLAIN), pb::Encoding::Gzip, None);
+ assert_eq!(message.body(), PLAIN);
+ assert!(!message.is_corrupted());
+ }
+
+ #[test]
+ fn undecodable_body_is_corrupted() {
+ let garbage = vec![0x1f, 0x8b, 0x00, 0x01, 0x02];
+ let message = view(garbage.clone(), pb::Encoding::Gzip, None);
+ assert_eq!(message.body(), garbage.as_slice());
+ assert!(message.is_corrupted());
+
+ let unknown_format = b"not compressed".to_vec();
+ let message = view(unknown_format.clone(), pb::Encoding::Gzip,
None);
+ assert_eq!(message.body(), unknown_format.as_slice());
+ assert!(message.is_corrupted());
+ }
+
+ #[test]
+ fn digest_is_verified_on_the_received_body() {
+ // The proxy computes the digest on the stored, i.e. still
compressed, body.
+ let compressed = zlib(PLAIN);
+ let message = view(
+ compressed.clone(),
+ pb::Encoding::Gzip,
+ digest(pb::DigestType::Md5, md5_hex(&compressed)),
+ );
+ assert_eq!(message.body(), PLAIN);
+ assert!(!message.is_corrupted());
+
+ let message = view(
+ compressed,
+ pb::Encoding::Gzip,
+ digest(pb::DigestType::Md5, md5_hex(PLAIN)),
+ );
+ assert!(message.is_corrupted());
+ }
+
+ #[test]
+ fn every_digest_type_is_verified() {
+ let crc32 = format!("{:X}", crc32fast::hash(PLAIN));
+ let sha1 = hex::encode_upper(
+ ring::digest::digest(&ring::digest::SHA1_FOR_LEGACY_USE_ONLY,
PLAIN).as_ref(),
+ );
+ for (r#type, checksum) in [
+ (pb::DigestType::Crc32, crc32),
+ (pb::DigestType::Md5, md5_hex(PLAIN)),
+ (pb::DigestType::Sha1, sha1),
+ ] {
+ let message = view(
+ PLAIN.to_vec(),
+ pb::Encoding::Identity,
+ digest(r#type, checksum.clone()),
+ );
+ assert!(!message.is_corrupted(), "{:?} {}", r#type, checksum);
+
+ let message = view(
+ PLAIN.to_vec(),
+ pb::Encoding::Identity,
+ digest(r#type, "0".to_string()),
+ );
+ assert!(message.is_corrupted(), "{:?}", r#type);
+ }
+ }
+ }
+
#[test]
fn common_test_message() {
let mut properties = HashMap::new();
diff --git a/rust/src/push_consumer.rs b/rust/src/push_consumer.rs
index b4c5fcff..d16c8d88 100644
--- a/rust/src/push_consumer.rs
+++ b/rust/src/push_consumer.rs
@@ -816,6 +816,20 @@ impl StandardConsumerWorker {
)
.await?;
for message in messages {
+ // Reference Java: `StandardConsumeService#consume` nacks a
corrupted message
+ // instead of passing it to the listener.
+ if message.is_corrupted() {
+ error!(
+ "Message is corrupted for standard consumption, prepare to
discard it, topic={}, message_id={}",
+ message.topic(),
+ message.message_id()
+ );
+ let delay =
retry_policy.get_next_attempt_delay(message.delivery_attempt());
+ ack_processor
+ .change_invisible_duration(message, delay)
+ .await;
+ continue;
+ }
let consume_result = consume_message(&self.message_listener,
&message);
match consume_result {
ConsumeResult::SUCCESS => {
@@ -958,6 +972,20 @@ impl FifoConsumerWorker {
retry_policy: &BackOffRetryPolicy,
) {
let max_delivery_attempts = retry_policy.get_max_attempts();
+ // Reference Java: `FifoConsumeService#getNextValidMessage` forwards a
corrupted message
+ // to the dead letter queue instead of passing it to the listener.
+ if message.is_corrupted() {
+ error!(
+ "Message is corrupted for FIFO consumption, prepare to discard
it, topic={}, message_id={}",
+ message.topic(),
+ message.message_id()
+ );
+ let delivery_attempt = message.delivery_attempt();
+ ack_processor
+ .forward_to_deadletter_queue(message, delivery_attempt,
max_delivery_attempts)
+ .await;
+ return;
+ }
loop {
let consume_result = consume_message(message_listener, &message);
match consume_result {
@@ -2663,4 +2691,140 @@ mod tests {
assert!(result.is_ok(), "{:?}", result);
assert_eq!(vec![2], *attempts.lock());
}
+
+ /// A message whose GZIP body can not be decompressed.
+ fn new_corrupted_message() -> pb::Message {
+ let mut message = new_message();
+ let system_properties = message.system_properties.as_mut().unwrap();
+ system_properties.body_encoding = pb::Encoding::Gzip as i32;
+ message.body = b"not compressed".to_vec();
+ message
+ }
+
+ fn new_listener_counter() -> (Arc<AtomicUsize>, Arc<MessageListener>) {
+ let count = Arc::new(AtomicUsize::new(0));
+ let counter = Arc::clone(&count);
+ let listener: Arc<MessageListener> = Arc::new(Box::new(move |_| {
+ counter.fetch_add(1, Ordering::Relaxed);
+ ConsumeResult::SUCCESS
+ }));
+ (count, listener)
+ }
+
+ #[tokio::test]
+ async fn test_standard_consume_discards_corrupted_message() {
+ let mut rpc_client = Session::default();
+ rpc_client.expect_receive_message().returning(|_| {
+ Ok(vec![ReceiveMessageResponse {
+ content: Some(Content::Message(new_corrupted_message())),
+ }])
+ });
+ let (listener_calls, listener) = new_listener_counter();
+ let consumer_worker = StandardConsumerWorker::new(rpc_client,
listener);
+ let retry_policy =
&BackOffRetryPolicy::Exponential(ExponentialBackOffRetryPolicy::new(
+ pb::ExponentialBackoff {
+ initial: Some(prost_types::Duration::from_str("1s").unwrap()),
+ max: Some(prost_types::Duration::from_str("60s").unwrap()),
+ multiplier: 2.0,
+ },
+ 3,
+ ));
+ let mut option = PushConsumerOption::default();
+ option.subscribe("test_topic", FilterExpression::new(FilterType::Tag,
"*"));
+ let mut session = Session::default();
+ session.expect_shadow_session().returning(Session::default);
+ session.expect_ack_message().never();
+ session
+ .expect_change_invisible_duration()
+ .times(1)
+ .returning(|_| {
+ Ok(ChangeInvisibleDurationResponse {
+ status: Some(Status {
+ code: Code::Ok as i32,
+ message: "".to_string(),
+ }),
+ receipt_handle: "".to_string(),
+ })
+ });
+ let mut ack_processor = AckEntryProcessor::new(
+ session,
+ Resource {
+ name: "test_group".to_string(),
+ resource_namespace: "".to_string(),
+ },
+ Resource {
+ name: "test_topic".to_string(),
+ resource_namespace: "".to_string(),
+ },
+ false,
+ );
+ let result = ConsumerWorker::Standard(consumer_worker)
+ .receive_messages(
+ &new_message_queue(),
+ &FilterExpression::sub_all(),
+ &option,
+ &mut ack_processor,
+ retry_policy,
+ )
+ .await;
+ assert!(result.is_ok(), "{:?}", result);
+ assert_eq!(0, listener_calls.load(Ordering::Relaxed));
+ }
+
+ #[tokio::test]
+ async fn test_fifo_consume_forwards_corrupted_message_to_dlq() {
+ let mut rpc_client = Session::default();
+ rpc_client.expect_receive_message().returning(|_| {
+ Ok(vec![ReceiveMessageResponse {
+ content: Some(Content::Message(new_corrupted_message())),
+ }])
+ });
+ let (listener_calls, listener) = new_listener_counter();
+ let consumer_worker = FifoConsumerWorker::new(rpc_client, listener);
+ let retry_policy =
&BackOffRetryPolicy::Customized(CustomizedBackOffRetryPolicy::new(
+ pb::CustomizedBackoff {
+ next: vec![prost_types::Duration::from_str("0.01s").unwrap()],
+ },
+ 1,
+ ));
+ let mut option = PushConsumerOption::default();
+ option.subscribe("test_topic", FilterExpression::new(FilterType::Tag,
"*"));
+ let mut session = Session::default();
+ session.expect_shadow_session().returning(Session::default);
+ session.expect_ack_message().never();
+ session
+ .expect_forward_to_deadletter_queue()
+ .times(1)
+ .returning(|_| {
+ Ok(ForwardMessageToDeadLetterQueueResponse {
+ status: Some(Status {
+ code: Code::Ok as i32,
+ message: "".to_string(),
+ }),
+ })
+ });
+ let mut ack_processor = AckEntryProcessor::new(
+ session,
+ Resource {
+ name: "test_group".to_string(),
+ resource_namespace: "".to_string(),
+ },
+ Resource {
+ name: "test_topic".to_string(),
+ resource_namespace: "".to_string(),
+ },
+ false,
+ );
+ let result = ConsumerWorker::Fifo(consumer_worker)
+ .receive_messages(
+ &new_message_queue(),
+ &FilterExpression::sub_all(),
+ &option,
+ &mut ack_processor,
+ retry_policy,
+ )
+ .await;
+ assert!(result.is_ok(), "{:?}", result);
+ assert_eq!(0, listener_calls.load(Ordering::Relaxed));
+ }
}
