From 633460b987f3c387f49394c287575974f305835c Mon Sep 17 00:00:00 2001 From: itxaiohanglover <1531137510@qq.com> Date: Wed, 22 Jul 2026 13:12:46 +0800 Subject: [PATCH 1/2] feat(java): support GZIP message body compression on producer side (#1288) Introduce ProducerBuilder#setCompressBodyThresholdBytes to enable opt-in GZIP compression for message bodies that reach the threshold. Compression is disabled by default, and the consumer side already decompresses transparently according to the body encoding. --- .../client/apis/producer/ProducerBuilder.java | 12 +++ .../impl/producer/ProducerBuilderImpl.java | 12 +++ .../impl/producer/PublishingSettings.java | 12 +++ .../java/message/PublishingMessageImpl.java | 17 +++- .../producer/ProducerBuilderImplTest.java | 12 +++ .../message/PublishingMessageImplTest.java | 79 +++++++++++++++++++ 6 files changed, 140 insertions(+), 4 deletions(-) create mode 100644 java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/producer/ProducerBuilder.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/producer/ProducerBuilder.java index cc4bdc6f0..3e9498ce7 100644 --- a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/producer/ProducerBuilder.java +++ b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/producer/ProducerBuilder.java @@ -74,6 +74,18 @@ public interface ProducerBuilder { */ ProducerBuilder setTransactionChecker(TransactionChecker checker); + /** + * Set the threshold in bytes to enable message body compression. + * + *

Compression is disabled by default. Once this threshold is set, a message whose body size reaches it is + * compressed with GZIP before sending, and the consumer decompresses the body transparently according to the + * body encoding, so no change is required on the consuming side. + * + * @param compressBodyThresholdBytes threshold in bytes to enable message body compression. + * @return the producer builder instance. + */ + ProducerBuilder setCompressBodyThresholdBytes(int compressBodyThresholdBytes); + /** * Finalize the build of {@link Producer} instance and start. * diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImpl.java index 30d4f9d54..ee2b44274 100644 --- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImpl.java +++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImpl.java @@ -38,6 +38,7 @@ public class ProducerBuilderImpl implements ProducerBuilder { private final Set topics = new HashSet<>(); private int maxAttempts = 3; private TransactionChecker checker = null; + private int compressBodyThresholdBytes = Integer.MAX_VALUE; public ProducerBuilderImpl() { } @@ -83,6 +84,16 @@ public ProducerBuilder setTransactionChecker(TransactionChecker checker) { return this; } + /** + * @see ProducerBuilder#setCompressBodyThresholdBytes(int) + */ + @Override + public ProducerBuilder setCompressBodyThresholdBytes(int compressBodyThresholdBytes) { + checkArgument(compressBodyThresholdBytes > 0, "compressBodyThresholdBytes should be positive"); + this.compressBodyThresholdBytes = compressBodyThresholdBytes; + return this; + } + /** * @see ProducerBuilder#build() */ @@ -90,6 +101,7 @@ public ProducerBuilder setTransactionChecker(TransactionChecker checker) { public Producer build() { checkNotNull(clientConfiguration, "clientConfiguration has not been set yet"); final ProducerImpl producer = new ProducerImpl(clientConfiguration, topics, maxAttempts, checker); + producer.publishingSettings.setCompressBodyThresholdBytes(compressBodyThresholdBytes); producer.startAsync().awaitRunning(); return producer; } diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java index f728c5342..1f725beb7 100644 --- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java +++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java @@ -45,6 +45,10 @@ public class PublishingSettings extends Settings { */ private volatile int maxBodySizeBytes = 4 * 1024 * 1024; private volatile boolean validateMessageType = true; + /** + * If message body size reaches this threshold, it is compressed with GZIP before sending. Disabled by default. + */ + private volatile int compressBodyThresholdBytes = Integer.MAX_VALUE; public PublishingSettings(String namespace, ClientId clientId, Endpoints accessPoint, ExponentialBackoffRetryPolicy retryPolicy, Duration requestTimeout, Set topics, @@ -57,6 +61,14 @@ public int getMaxBodySizeBytes() { return maxBodySizeBytes; } + public int getCompressBodyThresholdBytes() { + return compressBodyThresholdBytes; + } + + public void setCompressBodyThresholdBytes(int compressBodyThresholdBytes) { + this.compressBodyThresholdBytes = compressBodyThresholdBytes; + } + public boolean isValidateMessageType() { return validateMessageType; } diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java index 08b281e4e..2f9d049ef 100644 --- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java +++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java @@ -36,13 +36,22 @@ public class PublishingMessageImpl extends MessageImpl { private final MessageId messageId; private final MessageType messageType; + private final Encoding encoding; + private final byte[] transportBody; public PublishingMessageImpl(Message message, PublishingSettings publishingSettings, boolean txEnabled) throws IOException { super(message); - final int length = message.getBody().remaining(); + // Compress the message body if it reaches the compression threshold. + if (body.length >= publishingSettings.getCompressBodyThresholdBytes()) { + this.encoding = Encoding.GZIP; + this.transportBody = Utilities.compressBytesGZIP(body); + } else { + this.encoding = Encoding.IDENTITY; + this.transportBody = body; + } final int maxBodySizeBytes = publishingSettings.getMaxBodySizeBytes(); - if (length > maxBodySizeBytes) { + if (transportBody.length > maxBodySizeBytes) { throw new IOException("Message body size exceeds the threshold, max size=" + maxBodySizeBytes + " bytes"); } // Generate message id. @@ -114,7 +123,7 @@ public apache.rocketmq.v2.Message toProtobuf(String namespace, MessageQueueImpl // Born host .setBornHost(Utilities.hostName()) // Body encoding - .setBodyEncoding(Encoding.toProtobuf(Encoding.IDENTITY)) + .setBodyEncoding(Encoding.toProtobuf(encoding)) // Queue id .setQueueId(mq.getQueueId()) // Message type @@ -136,7 +145,7 @@ public apache.rocketmq.v2.Message toProtobuf(String namespace, MessageQueueImpl // Topic .setTopic(topicResource) // Message body - .setBody(ByteString.copyFrom(getBody())) + .setBody(ByteString.copyFrom(transportBody)) // System properties .setSystemProperties(systemProperties) // User properties diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImplTest.java index 9680a71d8..31f6549b4 100644 --- a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImplTest.java +++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/ProducerBuilderImplTest.java @@ -76,6 +76,18 @@ public void testSetTransactionChecker() { builder.setTransactionChecker(messageView -> TransactionResolution.COMMIT); } + @Test(expected = IllegalArgumentException.class) + public void testSetNegativeCompressBodyThresholdBytes() { + final ProducerBuilderImpl builder = new ProducerBuilderImpl(); + builder.setCompressBodyThresholdBytes(-1); + } + + @Test + public void testSetCompressBodyThresholdBytes() { + final ProducerBuilderImpl builder = new ProducerBuilderImpl(); + builder.setCompressBodyThresholdBytes(4096); + } + @Test(expected = NullPointerException.class) public void testBuildWithoutClientConfiguration() { final ProducerBuilderImpl builder = new ProducerBuilderImpl(); diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java new file mode 100644 index 000000000..fe6577fb0 --- /dev/null +++ b/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.client.java.message; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.apache.rocketmq.client.apis.message.Message; +import org.apache.rocketmq.client.java.impl.producer.PublishingSettings; +import org.apache.rocketmq.client.java.misc.Utilities; +import org.apache.rocketmq.client.java.route.MessageQueueImpl; +import org.apache.rocketmq.client.java.tool.TestBase; +import org.junit.Test; + +public class PublishingMessageImplTest extends TestBase { + + @Test + public void testBodyNotCompressedByDefault() throws IOException { + String topic = "testTopic"; + byte[] body = "foobar".getBytes(StandardCharsets.UTF_8); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.IDENTITY, pb.getSystemProperties().getBodyEncoding()); + assertArrayEquals(body, pb.getBody().toByteArray()); + } + + @Test + public void testBodyCompressedWhenThresholdReached() throws IOException { + String topic = "testTopic"; + byte[] body = new byte[4096]; + Arrays.fill(body, (byte) 'x'); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + settings.setCompressBodyThresholdBytes(1024); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.GZIP, pb.getSystemProperties().getBodyEncoding()); + final byte[] transportBody = pb.getBody().toByteArray(); + assertTrue(transportBody.length < body.length); + assertArrayEquals(body, Utilities.decompressBytes(transportBody)); + } + + @Test + public void testBodyNotCompressedBelowThreshold() throws IOException { + String topic = "testTopic"; + byte[] body = "foobar".getBytes(StandardCharsets.UTF_8); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + settings.setCompressBodyThresholdBytes(1024); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.IDENTITY, pb.getSystemProperties().getBodyEncoding()); + assertArrayEquals(body, pb.getBody().toByteArray()); + } +} From 621b7affaf5b68b0215387cf78b9ce6ea6ee504c Mon Sep 17 00:00:00 2001 From: itxaiohanglover <1531137510@qq.com> Date: Thu, 23 Jul 2026 10:24:39 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat(java):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20package-private=20setter,=20compression=20fallback,=20more?= =?UTF-8?q?=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Tighten PublishingSettings#setCompressBodyThresholdBytes to package-private and document it as a client-only setting (never synced from the server) - Fall back to the identity encoding when GZIP does not reduce the body size (already-compressed or encrypted payloads), avoiding transport inflation - Add tests: exact-threshold boundary, incompressible-body fallback, oversized incompressible body rejection, compressed-body-fits acceptance --- .../impl/producer/PublishingSettings.java | 3 +- .../java/message/PublishingMessageImpl.java | 18 +++-- .../PublishingSettingsTestHelper.java | 30 ++++++++ .../message/PublishingMessageImplTest.java | 68 ++++++++++++++++++- 4 files changed, 110 insertions(+), 9 deletions(-) create mode 100644 java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettingsTestHelper.java diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java index 1f725beb7..e67eb7bfe 100644 --- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java +++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettings.java @@ -65,7 +65,8 @@ public int getCompressBodyThresholdBytes() { return compressBodyThresholdBytes; } - public void setCompressBodyThresholdBytes(int compressBodyThresholdBytes) { + // Client-only setting: configured via ProducerBuilder and never synced from the server, see sync(). + void setCompressBodyThresholdBytes(int compressBodyThresholdBytes) { this.compressBodyThresholdBytes = compressBodyThresholdBytes; } diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java index 2f9d049ef..bd2ae7984 100644 --- a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java +++ b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java @@ -42,14 +42,20 @@ public class PublishingMessageImpl extends MessageImpl { public PublishingMessageImpl(Message message, PublishingSettings publishingSettings, boolean txEnabled) throws IOException { super(message); - // Compress the message body if it reaches the compression threshold. + // Compress the message body if it reaches the compression threshold. Fall back to the + // original body when compression does not reduce the size (e.g. already-compressed or + // encrypted payloads), which avoids inflating the transport body. + byte[] candidateBody = body; + Encoding candidateEncoding = Encoding.IDENTITY; if (body.length >= publishingSettings.getCompressBodyThresholdBytes()) { - this.encoding = Encoding.GZIP; - this.transportBody = Utilities.compressBytesGZIP(body); - } else { - this.encoding = Encoding.IDENTITY; - this.transportBody = body; + final byte[] compressedBody = Utilities.compressBytesGZIP(body); + if (compressedBody.length < body.length) { + candidateBody = compressedBody; + candidateEncoding = Encoding.GZIP; + } } + this.encoding = candidateEncoding; + this.transportBody = candidateBody; final int maxBodySizeBytes = publishingSettings.getMaxBodySizeBytes(); if (transportBody.length > maxBodySizeBytes) { throw new IOException("Message body size exceeds the threshold, max size=" + maxBodySizeBytes + " bytes"); diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettingsTestHelper.java b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettingsTestHelper.java new file mode 100644 index 000000000..bb55ead5c --- /dev/null +++ b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/producer/PublishingSettingsTestHelper.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.client.java.impl.producer; + +/** + * Test-only accessor for package-private mutators of {@link PublishingSettings}. + */ +public final class PublishingSettingsTestHelper { + private PublishingSettingsTestHelper() { + } + + public static void setCompressBodyThresholdBytes(PublishingSettings settings, int compressBodyThresholdBytes) { + settings.setCompressBodyThresholdBytes(compressBodyThresholdBytes); + } +} diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java index fe6577fb0..b6e11bdae 100644 --- a/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java +++ b/java/client/src/test/java/org/apache/rocketmq/client/java/message/PublishingMessageImplTest.java @@ -24,8 +24,10 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Random; import org.apache.rocketmq.client.apis.message.Message; import org.apache.rocketmq.client.java.impl.producer.PublishingSettings; +import org.apache.rocketmq.client.java.impl.producer.PublishingSettingsTestHelper; import org.apache.rocketmq.client.java.misc.Utilities; import org.apache.rocketmq.client.java.route.MessageQueueImpl; import org.apache.rocketmq.client.java.tool.TestBase; @@ -53,7 +55,7 @@ public void testBodyCompressedWhenThresholdReached() throws IOException { Arrays.fill(body, (byte) 'x'); final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); final PublishingSettings settings = fakeProducerSettings(); - settings.setCompressBodyThresholdBytes(1024); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); final MessageQueueImpl mq = fakeMessageQueueImpl(topic); final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); @@ -69,11 +71,73 @@ public void testBodyNotCompressedBelowThreshold() throws IOException { byte[] body = "foobar".getBytes(StandardCharsets.UTF_8); final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); final PublishingSettings settings = fakeProducerSettings(); - settings.setCompressBodyThresholdBytes(1024); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); final MessageQueueImpl mq = fakeMessageQueueImpl(topic); final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); assertEquals(apache.rocketmq.v2.Encoding.IDENTITY, pb.getSystemProperties().getBodyEncoding()); assertArrayEquals(body, pb.getBody().toByteArray()); } + + @Test + public void testBodyCompressedAtExactThreshold() throws IOException { + String topic = "testTopic"; + byte[] body = new byte[1024]; + Arrays.fill(body, (byte) 'x'); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.GZIP, pb.getSystemProperties().getBodyEncoding()); + assertArrayEquals(body, Utilities.decompressBytes(pb.getBody().toByteArray())); + } + + @Test + public void testBodyNotCompressedWhenGzipInflates() throws IOException { + String topic = "testTopic"; + // Incompressible pseudo-random body: GZIP output is larger than the input, + // so the implementation must fall back to the identity encoding. + byte[] body = new byte[4096]; + new Random(42).nextBytes(body); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.IDENTITY, pb.getSystemProperties().getBodyEncoding()); + assertArrayEquals(body, pb.getBody().toByteArray()); + } + + @Test(expected = IOException.class) + public void testIncompressibleBodyExceedingMaxSizeRejected() throws IOException { + String topic = "testTopic"; + // Larger than the default 4 MiB limit and incompressible: the transport body + // stays over the limit after the compression fallback, so it must be rejected. + byte[] body = new byte[5 * 1024 * 1024]; + new Random(42).nextBytes(body); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); + new PublishingMessageImpl(message, settings, false); + } + + @Test + public void testCompressibleBodyOverMaxSizeAcceptedWhenCompressedFits() throws IOException { + String topic = "testTopic"; + // Original body exceeds the 4 MiB limit, but the compressed body fits: the + // limit is validated against the transport (compressed) body by design. + byte[] body = new byte[5 * 1024 * 1024]; + Arrays.fill(body, (byte) 'x'); + final Message message = new MessageBuilderImpl().setTopic(topic).setBody(body).build(); + final PublishingSettings settings = fakeProducerSettings(); + PublishingSettingsTestHelper.setCompressBodyThresholdBytes(settings, 1024); + final PublishingMessageImpl publishingMessage = new PublishingMessageImpl(message, settings, false); + final MessageQueueImpl mq = fakeMessageQueueImpl(topic); + final apache.rocketmq.v2.Message pb = publishingMessage.toProtobuf(FAKE_NAMESPACE, mq); + assertEquals(apache.rocketmq.v2.Encoding.GZIP, pb.getSystemProperties().getBodyEncoding()); + assertArrayEquals(body, Utilities.decompressBytes(pb.getBody().toByteArray())); + } }