[ISSUE #8503] Add test cases for org.apache.rocketmq.common.chain/coldstr/compression/consumer

common/src/test/java/org/apache/rocketmq/common/chain/HandlerChainTest.java

@@ -0,0 +1,65 @@
+/*
+ * 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.chain;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class HandlerChainTest {
+
+    private HandlerChain<Integer, String> handlerChain;
+    private Handler<Integer, String> handler1;
+    private Handler<Integer, String> handler2;
+
+    @Before
+    public void setUp() {
+        handlerChain = HandlerChain.create();
+        handler1 = (t, chain) -> "Handler1";
+        handler2 = (t, chain) -> null;
+    }
+
+    @Test
+    public void testHandle_withEmptyChain() {
+        handlerChain.addNext(handler1);
+        handlerChain.handle(1);
+        assertNull("Expected null since the handler chain is empty", handlerChain.handle(2));
+    }
+
+    @Test
+    public void testHandle_withNonEmptyChain() {
+        handlerChain.addNext(handler1);
+
+        String result = handlerChain.handle(1);
+
+        assertEquals("Handler1", result);
+    }
+
+    @Test
+    public void testHandle_withMultipleHandlers() {
+        handlerChain.addNext(handler1);
+        handlerChain.addNext(handler2);
+
+        String result1 = handlerChain.handle(1);
+        String result2 = handlerChain.handle(2);
+
+        assertEquals("Handler1", result1);
+        assertNull("Expected null since there are no more handlers", result2);
+    }
+}

common/src/test/java/org/apache/rocketmq/common/coldctr/AccAndTimeStampTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.coldctr;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AccAndTimeStampTest {
+
+    private AccAndTimeStamp accAndTimeStamp;
+
+    @Before
+    public void setUp() {
+        accAndTimeStamp = new AccAndTimeStamp(new AtomicLong());
+    }
+
+    @Test
+    public void testInitialValues() {
+        assertEquals("Cold accumulator should be initialized to 0", 0, accAndTimeStamp.getColdAcc().get());
+        assertTrue("Last cold read time should be initialized to current time", accAndTimeStamp.getLastColdReadTimeMills() >= System.currentTimeMillis() - 1000);
+        assertTrue("Create time should be initialized to current time", accAndTimeStamp.getCreateTimeMills() >= System.currentTimeMillis() - 1000);
+    }
+
+    @Test
+    public void testSetColdAcc() {
+        AtomicLong newColdAcc = new AtomicLong(100L);
+        accAndTimeStamp.setColdAcc(newColdAcc);
+        assertEquals("Cold accumulator should be set to new value", newColdAcc, accAndTimeStamp.getColdAcc());
+    }
+
+    @Test
+    public void testSetLastColdReadTimeMills() {
+        long newLastColdReadTimeMills = System.currentTimeMillis() + 1000;
+        accAndTimeStamp.setLastColdReadTimeMills(newLastColdReadTimeMills);
+        assertEquals("Last cold read time should be set to new value", newLastColdReadTimeMills, accAndTimeStamp.getLastColdReadTimeMills().longValue());
+    }
+
+    @Test
+    public void testSetCreateTimeMills() {
+        long newCreateTimeMills = System.currentTimeMillis() + 2000;
+        accAndTimeStamp.setCreateTimeMills(newCreateTimeMills);
+        assertEquals("Create time should be set to new value", newCreateTimeMills, accAndTimeStamp.getCreateTimeMills().longValue());
+    }
+
+    @Test
+    public void testToStringContainsCorrectInformation() {
+        String toStringOutput = accAndTimeStamp.toString();
+        assertTrue("ToString should contain cold accumulator value", toStringOutput.contains("coldAcc=" + accAndTimeStamp.getColdAcc()));
+        assertTrue("ToString should contain last cold read time", toStringOutput.contains("lastColdReadTimeMills=" + accAndTimeStamp.getLastColdReadTimeMills()));
+        assertTrue("ToString should contain create time", toStringOutput.contains("createTimeMills=" + accAndTimeStamp.getCreateTimeMills()));
+    }
+}

common/src/test/java/org/apache/rocketmq/common/compression/CompressionTypeTest.java

@@ -0,0 +1,60 @@
+/*
+ * 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;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class CompressionTypeTest {
+
+    @Test
+    public void testCompressionTypeValues() {
+        assertEquals(1, CompressionType.LZ4.getValue(), "LZ4 value should be 1");
+        assertEquals(2, CompressionType.ZSTD.getValue(), "ZSTD value should be 2");
+        assertEquals(3, CompressionType.ZLIB.getValue(), "ZLIB value should be 3");
+    }
+
+    @Test
+    public void testCompressionTypeOf() {
+        assertEquals(CompressionType.LZ4, CompressionType.of("LZ4"), "CompressionType.of(LZ4) should return LZ4");
+        assertEquals(CompressionType.ZSTD, CompressionType.of("ZSTD"), "CompressionType.of(ZSTD) should return ZSTD");
+        assertEquals(CompressionType.ZLIB, CompressionType.of("ZLIB"), "CompressionType.of(ZLIB) should return ZLIB");
+
+        assertThrows(RuntimeException.class, () -> CompressionType.of("UNKNOWN"), "Unsupported compression type should throw RuntimeException");
+    }
+
+    @Test
+    public void testCompressionTypeFindByValue() {
+        assertEquals(CompressionType.LZ4, CompressionType.findByValue(1), "CompressionType.findByValue(1) should return LZ4");
+        assertEquals(CompressionType.ZSTD, CompressionType.findByValue(2), "CompressionType.findByValue(2) should return ZSTD");
+        assertEquals(CompressionType.ZLIB, CompressionType.findByValue(3), "CompressionType.findByValue(3) should return ZLIB");
+
+        assertEquals(CompressionType.ZLIB, CompressionType.findByValue(0), "CompressionType.findByValue(0) should return ZLIB for backward compatibility");
+
+        assertThrows(RuntimeException.class, () -> CompressionType.findByValue(99), "Invalid value should throw RuntimeException");
+    }
+
+    @Test
+    public void testCompressionFlag() {
+        assertEquals(MessageSysFlag.COMPRESSION_LZ4_TYPE, CompressionType.LZ4.getCompressionFlag(), "LZ4 compression flag is incorrect");
+        assertEquals(MessageSysFlag.COMPRESSION_ZSTD_TYPE, CompressionType.ZSTD.getCompressionFlag(), "ZSTD compression flag is incorrect");
+        assertEquals(MessageSysFlag.COMPRESSION_ZLIB_TYPE, CompressionType.ZLIB.getCompressionFlag(), "ZLIB compression flag is incorrect");
+    }
+}

common/src/test/java/org/apache/rocketmq/common/compression/CompressorFactoryTest.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 org.junit.Assert;
+import org.junit.Test;
+
+public class CompressorFactoryTest {
+
+    @Test
+    public void testGetCompressor_ReturnsNonNull() {
+        for (CompressionType type : CompressionType.values()) {
+            Compressor compressor = CompressorFactory.getCompressor(type);
+            Assert.assertNotNull("Compressor should not be null for type " + type, compressor);
+        }
+    }
+
+    @Test
+    public void testGetCompressor_ReturnsCorrectType() {
+        for (CompressionType type : CompressionType.values()) {
+            Compressor compressor = CompressorFactory.getCompressor(type);
+            Assert.assertTrue("Compressor type mismatch for " + type,
+                compressor instanceof Lz4Compressor && type == CompressionType.LZ4 ||
+                    compressor instanceof ZstdCompressor && type == CompressionType.ZSTD ||
+                    compressor instanceof ZlibCompressor && type == CompressionType.ZLIB);
+        }
+    }
+}

common/src/test/java/org/apache/rocketmq/common/compression/Lz4CompressorTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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 static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import org.junit.Test;
+
+public class Lz4CompressorTest {
+
+    private static final String TEST_STRING = "The quick brown fox jumps over the lazy dog";
+
+    @Test
+    public void testCompressAndDecompress() throws Exception {
+        byte[] originalData = TEST_STRING.getBytes();
+        Compressor compressor = new Lz4Compressor();
+        byte[] compressedData = compressor.compress(originalData, 1);
+        assertTrue("Compressed data should be bigger than original", compressedData.length > originalData.length);
+
+        byte[] decompressedData = compressor.decompress(compressedData);
+        assertArrayEquals("Decompressed data should match original data", originalData, decompressedData);
+    }
+
+    @Test
+    public void testCompressWithIOException() throws Exception {
+        byte[] originalData = new byte[] {1, 2, 3};
+        Compressor compressor = new Lz4Compressor();
+        compressor.compress(originalData, 1);
+    }
+
+    @Test(expected = IOException.class)
+    public void testDecompressWithIOException() throws Exception {
+        byte[] compressedData = new byte[] {1, 2, 3};
+        Compressor compressor = new Lz4Compressor();
+        compressor.decompress(compressedData);
+    }
+}

common/src/test/java/org/apache/rocketmq/common/compression/ZlibCompressorTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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 static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import org.junit.Test;
+
+public class ZlibCompressorTest {
+
+    private static final String TEST_STRING = "The quick brown fox jumps over the lazy dog";
+
+    @Test
+    public void testCompressionAndDecompression() throws Exception {
+        byte[] originalData = TEST_STRING.getBytes();
+        ZlibCompressor compressor = new ZlibCompressor();
+        byte[] compressedData = compressor.compress(originalData, 0);
+        assertTrue("Compressed data should be bigger than original", compressedData.length > originalData.length);
+
+        byte[] decompressedData = compressor.decompress(compressedData);
+        assertArrayEquals("Decompressed data should match original", originalData, decompressedData);
+    }
+
+    @Test
+    public void testCompressionFailureWithInvalidData() throws Exception {
+        byte[] originalData = new byte[] {0, 1, 2, 3, 4};
+        ZlibCompressor compressor = new ZlibCompressor();
+        compressor.compress(originalData, 0);
+    }
+
+    @Test(expected = IOException.class)
+    public void testDecompressionFailureWithInvalidData() throws Exception {
+        byte[] compressedData = new byte[] {0, 1, 2, 3, 4};
+        ZlibCompressor compressor = new ZlibCompressor();
+        compressor.decompress(compressedData); // Invalid compressed data
+    }
+}