diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java index b90a5b7908..e8d83efbc8 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java @@ -28,6 +28,8 @@ import org.apache.rocketmq.client.log.ClientLogger; import org.apache.rocketmq.client.producer.RequestFutureHolder; import org.apache.rocketmq.client.producer.RequestResponseFuture; import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.compression.Compressor; +import org.apache.rocketmq.common.compression.CompressorFactory; import org.apache.rocketmq.common.message.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.message.MessageDecoder; @@ -245,9 +247,11 @@ public class ClientRemotingProcessor extends AsyncNettyRequestProcessor implemen } byte[] body = request.getBody(); - if ((requestHeader.getSysFlag() & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) { + int sysFlag = requestHeader.getSysFlag(); + if ((sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) { try { - body = UtilAll.uncompress(body); + Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag)); + body = compressor.decompress(body); } catch (IOException e) { log.warn("err when uncompress constant", e); } diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java index 73569368f9..d08aac4e2b 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java @@ -67,7 +67,9 @@ import org.apache.rocketmq.client.producer.TransactionMQProducer; import org.apache.rocketmq.client.producer.TransactionSendResult; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.ServiceState; -import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.compression.CompressionType; +import org.apache.rocketmq.common.compression.Compressor; +import org.apache.rocketmq.common.compression.CompressorFactory; import org.apache.rocketmq.common.help.FAQUrl; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageAccessor; @@ -109,10 +111,14 @@ public class DefaultMQProducerImpl implements MQProducerInner { private ServiceState serviceState = ServiceState.CREATE_JUST; private MQClientInstance mQClientFactory; private ArrayList checkForbiddenHookList = new ArrayList(); - private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5")); private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy(); private ExecutorService asyncSenderExecutor; + // compression related + private int compressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5")); + private CompressionType compressType = CompressionType.of(System.getProperty(MixAll.MESSAGE_COMPRESS_TYPE, "ZLIB")); + private final Compressor compressor = CompressorFactory.getCompressor(compressType); + public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) { this(defaultMQProducer, null); } @@ -530,7 +536,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { } private void validateNameServerSetting() throws MQClientException { - List nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList(); + List nsList = this.getMqClientFactory().getMQClientAPIImpl().getNameServerAddressList(); if (null == nsList || nsList.isEmpty()) { throw new MQClientException( "No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION); @@ -724,6 +730,7 @@ public class DefaultMQProducerImpl implements MQProducerInner { boolean msgBodyCompressed = false; if (this.tryToCompressMessage(msg)) { sysFlag |= MessageSysFlag.COMPRESSED_FLAG; + sysFlag |= compressType.getCompressionFlag(); msgBodyCompressed = true; } @@ -886,20 +893,25 @@ public class DefaultMQProducerImpl implements MQProducerInner { throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null); } + public MQClientInstance getMqClientFactory() { + return mQClientFactory; + } + + @Deprecated public MQClientInstance getmQClientFactory() { return mQClientFactory; } private boolean tryToCompressMessage(final Message msg) { if (msg instanceof MessageBatch) { - //batch dose not support compressing right now + //batch does not support compressing right now return false; } byte[] body = msg.getBody(); if (body != null) { if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) { try { - byte[] data = UtilAll.compress(body, zipCompressLevel); + byte[] data = compressor.compress(body, compressLevel); if (data != null) { msg.setBody(data); return true; @@ -1561,16 +1573,16 @@ public class DefaultMQProducerImpl implements MQProducerInner { private void prepareSendRequest(final Message msg, long timeout) { String correlationId = CorrelationIdUtil.createCorrelationId(); - String requestClientId = this.getmQClientFactory().getClientId(); + String requestClientId = this.getMqClientFactory().getClientId(); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_CORRELATION_ID, correlationId); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, requestClientId); MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_TTL, String.valueOf(timeout)); - boolean hasRouteData = this.getmQClientFactory().getTopicRouteTable().containsKey(msg.getTopic()); + boolean hasRouteData = this.getMqClientFactory().getTopicRouteTable().containsKey(msg.getTopic()); if (!hasRouteData) { long beginTimestamp = System.currentTimeMillis(); this.tryToFindTopicPublishInfo(msg.getTopic()); - this.getmQClientFactory().sendHeartbeatToAllBrokerWithLock(); + this.getMqClientFactory().sendHeartbeatToAllBrokerWithLock(); long cost = System.currentTimeMillis() - beginTimestamp; if (cost > 500) { log.warn("prepare send request for <{}> cost {} ms", msg.getTopic(), cost); @@ -1582,12 +1594,20 @@ public class DefaultMQProducerImpl implements MQProducerInner { return topicPublishInfoTable; } - public int getZipCompressLevel() { - return zipCompressLevel; + public int getCompressLevel() { + return compressLevel; } - public void setZipCompressLevel(int zipCompressLevel) { - this.zipCompressLevel = zipCompressLevel; + public void setCompressLevel(int compressLevel) { + this.compressLevel = compressLevel; + } + + public CompressionType getCompressType() { + return compressType; + } + + public void setCompressType(CompressionType compressType) { + this.compressType = compressType; } public ServiceState getServiceState() { diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java index 19c276238a..86153f5263 100644 --- a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java +++ b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java @@ -438,7 +438,7 @@ public class AsyncTraceDispatcher implements TraceDispatcher { TopicPublishInfo topicPublishInfo = producer.getTopicPublishInfoTable().get(topic); if (null == topicPublishInfo || !topicPublishInfo.ok()) { producer.getTopicPublishInfoTable().putIfAbsent(topic, new TopicPublishInfo()); - producer.getmQClientFactory().updateTopicRouteInfoFromNameServer(topic); + producer.getMqClientFactory().updateTopicRouteInfoFromNameServer(topic); topicPublishInfo = producer.getTopicPublishInfoTable().get(topic); } if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) { diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionTraceHookImpl.java b/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionTraceHookImpl.java index cbd755ba39..8172d0459d 100644 --- a/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionTraceHookImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionTraceHookImpl.java @@ -63,7 +63,7 @@ public class EndTransactionTraceHookImpl implements EndTransactionHook { traceBean.setKeys(context.getMessage().getKeys()); traceBean.setStoreHost(context.getBrokerAddr()); traceBean.setMsgType(MessageType.Trans_msg_Commit); - traceBean.setClientHost(((AsyncTraceDispatcher)localDispatcher).getHostProducer().getmQClientFactory().getClientId()); + traceBean.setClientHost(((AsyncTraceDispatcher)localDispatcher).getHostProducer().getMqClientFactory().getClientId()); traceBean.setMsgId(context.getMsgId()); traceBean.setTransactionState(context.getTransactionState()); traceBean.setTransactionId(context.getTransactionId()); diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java index b97b30e08f..5ce761c30f 100644 --- a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java @@ -102,7 +102,7 @@ public class DefaultMQProducerTest { field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod(); @@ -358,7 +358,7 @@ public class DefaultMQProducerTest { producer.setCallbackExecutor(customized); NettyRemotingClient remotingClient = (NettyRemotingClient) producer.getDefaultMQProducerImpl() - .getmQClientFactory().getMQClientAPIImpl().getRemotingClient(); + .getMqClientFactory().getMQClientAPIImpl().getRemotingClient(); assertThat(remotingClient.getCallbackExecutor()).isEqualTo(customized); } diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java index f7771ec597..ee94b90df2 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java @@ -217,7 +217,7 @@ public class DefaultMQConsumerWithTraceTest { @Test public void testPullMessage_WithTrace_Success() throws InterruptedException, RemotingException, MQBrokerException, MQClientException { - traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); + traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); final CountDownLatch countDownLatch = new CountDownLatch(1); final AtomicReference messageAtomic = new AtomicReference(); diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java index 3f7031e8fe..884a2f6880 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java @@ -218,7 +218,7 @@ public class DefaultMQLitePullConsumerWithTraceTest { field.setAccessible(true); field.set(litePullConsumerImpl, offsetStore); - traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); + traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(PullCallback.class))) diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithOpenTracingTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithOpenTracingTest.java index 5d64a93f03..cc57a03164 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithOpenTracingTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithOpenTracingTest.java @@ -100,7 +100,7 @@ public class DefaultMQProducerWithOpenTracingTest { field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod(); @@ -112,7 +112,7 @@ public class DefaultMQProducerWithOpenTracingTest { @Test public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException { - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute()); producer.send(message); assertThat(tracer.finishedSpans().size()).isEqualTo(1); diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java index 234e32e680..b951ae88de 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java @@ -109,7 +109,7 @@ public class DefaultMQProducerWithTraceTest { field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod(); @@ -121,7 +121,7 @@ public class DefaultMQProducerWithTraceTest { @Test public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException { - traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); + traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute()); final CountDownLatch countDownLatch = new CountDownLatch(1); try { diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithOpenTracingTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithOpenTracingTest.java index dd6d1083ce..5de15f451e 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithOpenTracingTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithOpenTracingTest.java @@ -118,7 +118,7 @@ public class TransactionMQProducerWithOpenTracingTest { field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod(); @@ -130,7 +130,7 @@ public class TransactionMQProducerWithOpenTracingTest { @Test public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException { - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, producer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute()); producer.sendMessageInTransaction(message, null); diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java index a290501755..3454bf078c 100644 --- a/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/trace/TransactionMQProducerWithTraceTest.java @@ -128,7 +128,7 @@ public class TransactionMQProducerWithTraceTest { field.setAccessible(true); field.set(mQClientFactory, mQClientAPIImpl); - producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); + producer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl()); Field fieldHooks = DefaultMQProducerImpl.class.getDeclaredField("endTransactionHookList"); fieldHooks.setAccessible(true); @@ -146,7 +146,7 @@ public class TransactionMQProducerWithTraceTest { @Test public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException { - traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); + traceProducer.getDefaultMQProducerImpl().getMqClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl()); when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute()); final AtomicReference context = new AtomicReference(); doAnswer(new Answer() { diff --git a/common/pom.xml b/common/pom.xml index 610be62347..c5a6689011 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -40,5 +40,13 @@ commons-validator commons-validator + + com.github.luben + zstd-jni + + + org.lz4 + lz4-java + diff --git a/common/src/main/java/org/apache/rocketmq/common/MixAll.java b/common/src/main/java/org/apache/rocketmq/common/MixAll.java index 5f78174486..05658d0d64 100644 --- a/common/src/main/java/org/apache/rocketmq/common/MixAll.java +++ b/common/src/main/java/org/apache/rocketmq/common/MixAll.java @@ -50,6 +50,7 @@ public class MixAll { public static final String ROCKETMQ_HOME_PROPERTY = "rocketmq.home.dir"; public static final String NAMESRV_ADDR_ENV = "NAMESRV_ADDR"; public static final String NAMESRV_ADDR_PROPERTY = "rocketmq.namesrv.addr"; + public static final String MESSAGE_COMPRESS_TYPE = "rocketmq.message.compressType"; public static final String MESSAGE_COMPRESS_LEVEL = "rocketmq.message.compressLevel"; public static final String DEFAULT_NAMESRV_ADDR_LOOKUP = "jmenv.tbsite.net"; public static final String WS_DOMAIN_NAME = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP); diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java index a15b4fadf6..feeb6c714e 100644 --- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java +++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java @@ -295,6 +295,10 @@ public class UtilAll { return (byte) "0123456789ABCDEF".indexOf(c); } + /** + * use {@link org.apache.rocketmq.common.compression.Compressor#decompress(byte[])} instead. + */ + @Deprecated public static byte[] uncompress(final byte[] src) throws IOException { byte[] result = src; byte[] uncompressData = new byte[src.length]; @@ -335,6 +339,10 @@ public class UtilAll { return result; } + /** + * use {@link org.apache.rocketmq.common.compression.Compressor#compress(byte[], int)} instead. + */ + @Deprecated public static byte[] compress(final byte[] src, final int level) throws IOException { byte[] result = src; ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length); diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/CompressionType.java b/common/src/main/java/org/apache/rocketmq/common/compression/CompressionType.java new file mode 100644 index 0000000000..d748fc47a2 --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/CompressionType.java @@ -0,0 +1,90 @@ +/* + * 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.common.compression; + +import org.apache.rocketmq.common.sysflag.MessageSysFlag; + +public enum CompressionType { + + /** + * Compression types number can be extended to seven {@link MessageSysFlag} + * + * Benchmarks from https://github.com/facebook/zstd + * + * | Compressor | Ratio | Compression | Decompress | + * |----------------|---------|-------------|------------| + * | zstd 1.5.1 | 2.887 | 530 MB/s | 1700 MB/s | + * | zlib 1.2.11 | 2.743 | 95 MB/s | 400 MB/s | + * | lz4 1.9.3 | 2.101 | 740 MB/s | 4500 MB/s | + * + */ + + LZ4(1), + ZSTD(2), + ZLIB(3); + + private final int value; + + CompressionType(int value) { + this.value = value; + } + + public int getValue() { + return value; + } + + public static CompressionType of(String name) { + switch (name.trim().toUpperCase()) { + case "LZ4": + return CompressionType.LZ4; + case "ZSTD": + return CompressionType.ZSTD; + case "ZLIB": + return CompressionType.ZLIB; + default: + throw new RuntimeException("Unsupported compress type name: " + name); + } + } + + public static CompressionType findByValue(int value) { + switch (value) { + case 1: + return LZ4; + case 2: + return ZSTD; + case 0: // To be compatible for older versions without compression type + case 3: + return ZLIB; + default: + throw new RuntimeException("Unknown compress type value: " + value); + } + } + + public int getCompressionFlag() { + switch (value) { + case 1: + return MessageSysFlag.COMPRESSION_LZ4_TYPE; + case 2: + return MessageSysFlag.COMPRESSION_ZSTD_TYPE; + case 3: + return MessageSysFlag.COMPRESSION_ZLIB_TYPE; + default: + throw new RuntimeException("Unsupported compress type flag: " + value); + } + } +} diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/Compressor.java b/common/src/main/java/org/apache/rocketmq/common/compression/Compressor.java new file mode 100644 index 0000000000..c900045b24 --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/Compressor.java @@ -0,0 +1,42 @@ +/* + * 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.common.compression; + +import java.io.IOException; + +public interface Compressor { + + /** + * Compress message by different compressor. + * + * @param src bytes ready to compress + * @param level compression level used to balance compression rate and time consumption + * @return compressed byte data + * @throws IOException + */ + byte[] compress(byte[] src, int level) throws IOException; + + /** + * Decompress message by different compressor. + * + * @param src bytes ready to decompress + * @return decompressed byte data + * @throws IOException + */ + byte[] decompress(byte[] src) throws IOException; +} diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/CompressorFactory.java b/common/src/main/java/org/apache/rocketmq/common/compression/CompressorFactory.java new file mode 100644 index 0000000000..aace1d956a --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/CompressorFactory.java @@ -0,0 +1,36 @@ +/* + * 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.common.compression; + +import java.util.EnumMap; + +public class CompressorFactory { + private static final EnumMap COMPRESSORS; + + static { + COMPRESSORS = new EnumMap<>(CompressionType.class); + COMPRESSORS.put(CompressionType.LZ4, new Lz4Compressor()); + COMPRESSORS.put(CompressionType.ZSTD, new ZstdCompressor()); + COMPRESSORS.put(CompressionType.ZLIB, new ZlibCompressor()); + } + + public static Compressor getCompressor(CompressionType type) { + return COMPRESSORS.get(type); + } + +} diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/Lz4Compressor.java b/common/src/main/java/org/apache/rocketmq/common/compression/Lz4Compressor.java new file mode 100644 index 0000000000..4e537d856c --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/Lz4Compressor.java @@ -0,0 +1,86 @@ +/* + * 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.common.compression; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import net.jpountz.lz4.LZ4FrameInputStream; +import net.jpountz.lz4.LZ4FrameOutputStream; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; + +public class Lz4Compressor implements Compressor { + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME); + + @Override + public byte[] compress(byte[] src, int level) throws IOException { + byte[] result = src; + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length); + LZ4FrameOutputStream outputStream = new LZ4FrameOutputStream(byteArrayOutputStream); + try { + outputStream.write(src); + outputStream.flush(); + outputStream.close(); + result = byteArrayOutputStream.toByteArray(); + } catch (IOException e) { + log.error("Failed to compress data by lz4", e); + throw e; + } finally { + try { + byteArrayOutputStream.close(); + } catch (IOException ignored) { + } + } + return result; + } + + @Override + public byte[] decompress(byte[] src) throws IOException { + byte[] result = src; + byte[] uncompressData = new byte[src.length]; + ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src); + LZ4FrameInputStream lz4InputStream = new LZ4FrameInputStream(byteArrayInputStream); + ByteArrayOutputStream resultOutputStream = new ByteArrayOutputStream(src.length); + + try { + while (true) { + int len = lz4InputStream.read(uncompressData, 0, uncompressData.length); + if (len <= 0) { + break; + } + resultOutputStream.write(uncompressData, 0, len); + } + resultOutputStream.flush(); + resultOutputStream.close(); + result = resultOutputStream.toByteArray(); + } catch (IOException e) { + throw e; + } finally { + try { + lz4InputStream.close(); + byteArrayInputStream.close(); + } catch (IOException e) { + log.warn("Failed to close the lz4 compress stream ", e); + } + } + + return result; + } +} diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/ZlibCompressor.java b/common/src/main/java/org/apache/rocketmq/common/compression/ZlibCompressor.java new file mode 100644 index 0000000000..64e798f03a --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/ZlibCompressor.java @@ -0,0 +1,98 @@ +/* + * 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.common.compression; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.zip.DeflaterOutputStream; +import java.util.zip.InflaterInputStream; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; + +public class ZlibCompressor implements Compressor { + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME); + + @Override + public byte[] compress(byte[] src, int level) throws IOException { + byte[] result = src; + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length); + java.util.zip.Deflater defeater = new java.util.zip.Deflater(level); + DeflaterOutputStream deflaterOutputStream = new DeflaterOutputStream(byteArrayOutputStream, defeater); + try { + deflaterOutputStream.write(src); + deflaterOutputStream.finish(); + deflaterOutputStream.close(); + result = byteArrayOutputStream.toByteArray(); + } catch (IOException e) { + defeater.end(); + throw e; + } finally { + try { + byteArrayOutputStream.close(); + } catch (IOException ignored) { + } + + defeater.end(); + } + + return result; + } + + @Override + public byte[] decompress(byte[] src) throws IOException { + byte[] result = src; + byte[] uncompressData = new byte[src.length]; + ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src); + InflaterInputStream inflaterInputStream = new InflaterInputStream(byteArrayInputStream); + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length); + + try { + while (true) { + int len = inflaterInputStream.read(uncompressData, 0, uncompressData.length); + if (len <= 0) { + break; + } + byteArrayOutputStream.write(uncompressData, 0, len); + } + byteArrayOutputStream.flush(); + result = byteArrayOutputStream.toByteArray(); + } catch (IOException e) { + throw e; + } finally { + try { + byteArrayInputStream.close(); + } catch (IOException e) { + log.error("Failed to close the stream", e); + } + try { + inflaterInputStream.close(); + } catch (IOException e) { + log.error("Failed to close the stream", e); + } + try { + byteArrayOutputStream.close(); + } catch (IOException e) { + log.error("Failed to close the stream", e); + } + } + + return result; + } +} diff --git a/common/src/main/java/org/apache/rocketmq/common/compression/ZstdCompressor.java b/common/src/main/java/org/apache/rocketmq/common/compression/ZstdCompressor.java new file mode 100644 index 0000000000..3a220ce9fa --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/compression/ZstdCompressor.java @@ -0,0 +1,86 @@ +/* + * 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.common.compression; + +import com.github.luben.zstd.ZstdInputStream; +import com.github.luben.zstd.ZstdOutputStream; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; + +public class ZstdCompressor implements Compressor { + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME); + + @Override + public byte[] compress(byte[] src, int level) throws IOException { + byte[] result = src; + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(src.length); + ZstdOutputStream outputStream = new ZstdOutputStream(byteArrayOutputStream, level); + try { + outputStream.write(src); + outputStream.flush(); + outputStream.close(); + result = byteArrayOutputStream.toByteArray(); + } catch (IOException e) { + log.error("Failed to compress data by zstd", e); + throw e; + } finally { + try { + byteArrayOutputStream.close(); + } catch (IOException ignored) { + } + } + return result; + } + + @Override + public byte[] decompress(byte[] src) throws IOException { + byte[] result = src; + byte[] uncompressData = new byte[src.length]; + ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src); + ZstdInputStream zstdInputStream = new ZstdInputStream(byteArrayInputStream); + ByteArrayOutputStream resultOutputStream = new ByteArrayOutputStream(src.length); + + try { + while (true) { + int len = zstdInputStream.read(uncompressData, 0, uncompressData.length); + if (len <= 0) { + break; + } + resultOutputStream.write(uncompressData, 0, len); + } + resultOutputStream.flush(); + resultOutputStream.close(); + result = resultOutputStream.toByteArray(); + } catch (IOException e) { + throw e; + } finally { + try { + zstdInputStream.close(); + byteArrayInputStream.close(); + } catch (IOException e) { + log.warn("Failed to close the zstd compress stream", e); + } + } + + return result; + } +} \ No newline at end of file diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java index 30cee69c2d..7612ef4043 100644 --- a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java +++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java @@ -29,6 +29,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.compression.Compressor; +import org.apache.rocketmq.common.compression.CompressorFactory; import org.apache.rocketmq.common.sysflag.MessageSysFlag; public class MessageDecoder { @@ -163,7 +165,8 @@ public class MessageDecoder { int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20; byte[] newBody = messageExt.getBody(); if (needCompress && (sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) { - newBody = UtilAll.compress(body, 5); + Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag)); + newBody = compressor.compress(body, 5); } int bodyLength = newBody.length; int storeSize = messageExt.getStoreSize(); @@ -347,7 +350,8 @@ public class MessageDecoder { // uncompress body if (deCompressBody && (sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) { - body = UtilAll.uncompress(body); + Compressor compressor = CompressorFactory.getCompressor(MessageSysFlag.getCompressionType(sysFlag)); + body = compressor.decompress(body); } msgExt.setBody(body); diff --git a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java b/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java index d534571eb5..822478a83c 100644 --- a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java +++ b/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java @@ -16,7 +16,21 @@ */ package org.apache.rocketmq.common.sysflag; +import org.apache.rocketmq.common.compression.CompressionType; + public class MessageSysFlag { + + /** + * Meaning of each bit in the system flag + * + * | bit | 7 | 6 | 5 | 4 | 3 | 2 | 1 | 0 | + * |--------|---|---|-----------|----------|-------------|------------------|------------------|------------------| + * | byte 1 | | | STOREHOST | BORNHOST | TRANSACTION | TRANSACTION | MULTI_TAGS | COMPRESSED | + * | byte 2 | | | | | | COMPRESSION_TYPE | COMPRESSION_TYPE | COMPRESSION_TYPE | + * | byte 3 | | | | | | | | | + * | byte 4 | | | | | | | | | + * + */ public final static int COMPRESSED_FLAG = 0x1; public final static int MULTI_TAGS_FLAG = 0x1 << 1; public final static int TRANSACTION_NOT_TYPE = 0; @@ -26,6 +40,12 @@ public class MessageSysFlag { public final static int BORNHOST_V6_FLAG = 0x1 << 4; public final static int STOREHOSTADDRESS_V6_FLAG = 0x1 << 5; + // COMPRESSION_TYPE + public final static int COMPRESSION_LZ4_TYPE = 0x1 << 8; + public final static int COMPRESSION_ZSTD_TYPE = 0x2 << 8; + public final static int COMPRESSION_ZLIB_TYPE = 0x3 << 8; + public final static int COMPRESSION_TYPE_COMPARATOR = 0x7 << 8; + public static int getTransactionValue(final int flag) { return flag & TRANSACTION_ROLLBACK_TYPE; } @@ -38,4 +58,9 @@ public class MessageSysFlag { return flag & (~COMPRESSED_FLAG); } + // To match the compression type + public static CompressionType getCompressionType(final int flag) { + return CompressionType.findByValue((flag & COMPRESSION_TYPE_COMPARATOR) >> 8); + } + } diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTest.java new file mode 100644 index 0000000000..8110cfb6e3 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTest.java @@ -0,0 +1,106 @@ +/* + * 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.common.compression; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Random; +import org.apache.commons.lang3.RandomStringUtils; +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CompressionTest { + + private int level; + private Compressor zstd; + private Compressor zlib; + private Compressor lz4; + + @Before + public void setUp() { + level = 5; + zstd = CompressorFactory.getCompressor(CompressionType.ZSTD); + zlib = CompressorFactory.getCompressor(CompressionType.ZLIB); + lz4 = CompressorFactory.getCompressor(CompressionType.LZ4); + } + + @Test + public void testCompressionZlib() throws IOException { + assertThat(CompressionType.of("zlib")).isEqualTo(CompressionType.ZLIB); + assertThat(CompressionType.of(" ZLiB ")).isEqualTo(CompressionType.ZLIB); + assertThat(CompressionType.of("ZLIB")).isEqualTo(CompressionType.ZLIB); + + int randomKB = 4096; + Random random = new Random(); + for (int i = 0; i < 5; ++i) { + String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024); + byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8); + byte[] compressed = zlib.compress(srcBytes, level); + byte[] decompressed = zlib.decompress(compressed); + // compression ratio may be negative for some random string data + // assertThat(compressed.length).isLessThan(srcBytes.length); + assertThat(decompressed).isEqualTo(srcBytes); + assertThat(new String(decompressed)).isEqualTo(message); + } + } + + @Test + public void testCompressionZstd() throws IOException { + assertThat(CompressionType.of("zstd")).isEqualTo(CompressionType.ZSTD); + assertThat(CompressionType.of("ZStd ")).isEqualTo(CompressionType.ZSTD); + assertThat(CompressionType.of("ZSTD")).isEqualTo(CompressionType.ZSTD); + + int randomKB = 4096; + Random random = new Random(); + for (int i = 0; i < 5; ++i) { + String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024); + byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8); + byte[] compressed = zstd.compress(srcBytes, level); + byte[] decompressed = zstd.decompress(compressed); + // compression ratio may be negative for some random string data + // assertThat(compressed.length).isLessThan(srcBytes.length); + assertThat(decompressed).isEqualTo(srcBytes); + assertThat(new String(decompressed)).isEqualTo(message); + } + } + + @Test + public void testCompressionLz4() throws IOException { + assertThat(CompressionType.of("lz4")).isEqualTo(CompressionType.LZ4); + + int randomKB = 4096; + Random random = new Random(); + for (int i = 0; i < 5; ++i) { + String message = RandomStringUtils.randomAlphanumeric(random.nextInt(randomKB) * 1024); + byte[] srcBytes = message.getBytes(StandardCharsets.UTF_8); + byte[] compressed = lz4.compress(srcBytes, level); + byte[] decompressed = lz4.decompress(compressed); + // compression ratio may be negative for some random string data + // assertThat(compressed.length).isLessThan(srcBytes.length); + assertThat(decompressed).isEqualTo(srcBytes); + assertThat(new String(decompressed)).isEqualTo(message); + } + } + + @Test(expected = RuntimeException.class) + public void testCompressionUnsupportedType() { + CompressionType.of("snappy"); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/sysflag/CompressionFlagTest.java b/common/src/test/java/org/apache/rocketmq/common/sysflag/CompressionFlagTest.java new file mode 100644 index 0000000000..63906c321d --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/sysflag/CompressionFlagTest.java @@ -0,0 +1,55 @@ +/* + * 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.common.sysflag; + +import org.apache.rocketmq.common.compression.CompressionType; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CompressionFlagTest { + + @Test + public void testCompressionFlag() { + int flag = 0; + flag |= MessageSysFlag.COMPRESSED_FLAG; + + assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZLIB); + + flag |= MessageSysFlag.COMPRESSION_LZ4_TYPE; + assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.LZ4); + + flag &= (~MessageSysFlag.COMPRESSION_TYPE_COMPARATOR); + flag |= MessageSysFlag.COMPRESSION_ZSTD_TYPE; + assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZSTD); + + + flag &= (~MessageSysFlag.COMPRESSION_TYPE_COMPARATOR); + flag |= MessageSysFlag.COMPRESSION_ZLIB_TYPE; + assertThat(MessageSysFlag.getCompressionType(flag)).isEqualTo(CompressionType.ZLIB); + } + + @Test(expected = RuntimeException.class) + public void testCompressionFlagNotMatch() { + int flag = 0; + flag |= MessageSysFlag.COMPRESSED_FLAG; + flag |= (0x4 << 8); + + MessageSysFlag.getCompressionType(flag); + } +} diff --git a/pom.xml b/pom.xml index ec051303dd..c8535c0f02 100644 --- a/pom.xml +++ b/pom.xml @@ -605,8 +605,16 @@ commons-validator 1.7 - - + + com.github.luben + zstd-jni + 1.5.2-2 + + + org.lz4 + lz4-java + 1.8.0 +