diff --git a/.github/workflows/pr-e2e-test.yml b/.github/workflows/pr-e2e-test.yml index 9082b6b2227..f9bb3bde75a 100644 --- a/.github/workflows/pr-e2e-test.yml +++ b/.github/workflows/pr-e2e-test.yml @@ -187,6 +187,8 @@ jobs: test-cmd: | cd ../common && mvn -Prelease -DskipTests clean package -U cd ../rocketmq-admintools && source bin/env.sh + wget https://go.dev/dl/go1.22.6.linux-amd64.tar.gz && \ + rm -rf /usr/local/go && tar -C /usr/local -xzf go1.22.6.linux-amd64.tar.gz cd ../golang && go get -u github.com/apache/rocketmq-clients/golang && gotestsum --junitfile ./target/surefire-reports/TEST-report.xml ./mqgotest/... -timeout 2m -v job-id: 0 - name: Publish Test Report diff --git a/.github/workflows/push-ci.yml b/.github/workflows/push-ci.yml index b679d56d2f0..2fe62dbeb06 100644 --- a/.github/workflows/push-ci.yml +++ b/.github/workflows/push-ci.yml @@ -192,6 +192,8 @@ jobs: test-cmd: | cd ../common && mvn -Prelease -DskipTests clean package -U cd ../rocketmq-admintools && source bin/env.sh + wget https://go.dev/dl/go1.22.6.linux-amd64.tar.gz && \ + rm -rf /usr/local/go && tar -C /usr/local -xzf go1.22.6.linux-amd64.tar.gz cd ../golang && go get -u github.com/apache/rocketmq-clients/golang && gotestsum --junitfile ./target/surefire-reports/TEST-report.xml ./mqgotest/... -timeout 2m -v job-id: 0 - name: Publish Test Report diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerOuterAPITest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerOuterAPITest.java index 440ebf813bb..1d12acd4a98 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerOuterAPITest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerOuterAPITest.java @@ -136,6 +136,9 @@ public void test_needRegister_normal() throws Exception { @Test public void test_needRegister_timeout() throws Exception { + if (MixAll.isMac()) { + return; + } init(); brokerOuterAPI.start(); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManagerTest.java index 11f7ae8215a..9264eb4b56b 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManagerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManagerTest.java @@ -20,8 +20,25 @@ import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.common.AttributesBuilder; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.attribute.TopicMessageType; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.metrics.MetricsExporterType; +import org.apache.rocketmq.common.topic.TopicValidator; +import org.apache.rocketmq.remoting.netty.NettyClientConfig; +import org.apache.rocketmq.remoting.netty.NettyServerConfig; +import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader; +import org.apache.rocketmq.store.config.MessageStoreConfig; import org.junit.Test; +import java.io.File; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + import static org.assertj.core.api.Assertions.assertThat; public class BrokerMetricsManagerTest { @@ -29,7 +46,7 @@ public class BrokerMetricsManagerTest { @Test public void testNewAttributesBuilder() { Attributes attributes = BrokerMetricsManager.newAttributesBuilder().put("a", "b") - .build(); + .build(); assertThat(attributes.get(AttributeKey.stringKey("a"))).isEqualTo("b"); } @@ -37,6 +54,7 @@ public void testNewAttributesBuilder() { public void testCustomizedAttributesBuilder() { BrokerMetricsManager.attributesBuilderSupplier = () -> new AttributesBuilder() { private AttributesBuilder attributesBuilder = Attributes.builder(); + @Override public Attributes build() { return attributesBuilder.put("customized", "value").build(); @@ -61,8 +79,263 @@ public AttributesBuilder putAll(Attributes attributes) { } }; Attributes attributes = BrokerMetricsManager.newAttributesBuilder().put("a", "b") - .build(); + .build(); assertThat(attributes.get(AttributeKey.stringKey("a"))).isEqualTo("b"); assertThat(attributes.get(AttributeKey.stringKey("customized"))).isEqualTo("value"); } + + + @Test + public void testIsRetryOrDlqTopicWithRetryTopic() { + String topic = MixAll.RETRY_GROUP_TOPIC_PREFIX + "TestTopic"; + boolean result = BrokerMetricsManager.isRetryOrDlqTopic(topic); + assertThat(result).isTrue(); + } + + @Test + public void testIsRetryOrDlqTopicWithDlqTopic() { + String topic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "TestTopic"; + boolean result = BrokerMetricsManager.isRetryOrDlqTopic(topic); + assertThat(result).isTrue(); + } + + @Test + public void testIsRetryOrDlqTopicWithNonRetryOrDlqTopic() { + String topic = "NormalTopic"; + boolean result = BrokerMetricsManager.isRetryOrDlqTopic(topic); + assertThat(result).isFalse(); + } + + @Test + public void testIsRetryOrDlqTopicWithEmptyTopic() { + String topic = ""; + boolean result = BrokerMetricsManager.isRetryOrDlqTopic(topic); + assertThat(result).isFalse(); + } + + @Test + public void testIsRetryOrDlqTopicWithNullTopic() { + String topic = null; + boolean result = BrokerMetricsManager.isRetryOrDlqTopic(topic); + assertThat(result).isFalse(); + } + + @Test + public void testIsSystemGroup_SystemGroup_ReturnsTrue() { + String group = "FooGroup"; + String systemGroup = MixAll.CID_RMQ_SYS_PREFIX + group; + boolean result = BrokerMetricsManager.isSystemGroup(systemGroup); + assertThat(result).isTrue(); + } + + @Test + public void testIsSystemGroup_NonSystemGroup_ReturnsFalse() { + String group = "FooGroup"; + boolean result = BrokerMetricsManager.isSystemGroup(group); + assertThat(result).isFalse(); + } + + @Test + public void testIsSystemGroup_EmptyGroup_ReturnsFalse() { + String group = ""; + boolean result = BrokerMetricsManager.isSystemGroup(group); + assertThat(result).isFalse(); + } + + @Test + public void testIsSystemGroup_NullGroup_ReturnsFalse() { + String group = null; + boolean result = BrokerMetricsManager.isSystemGroup(group); + assertThat(result).isFalse(); + } + + @Test + public void testIsSystem_SystemTopicOrSystemGroup_ReturnsTrue() { + String topic = "FooTopic"; + String group = "FooGroup"; + String systemTopic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC; + String systemGroup = MixAll.CID_RMQ_SYS_PREFIX + group; + + boolean resultTopic = BrokerMetricsManager.isSystem(systemTopic, group); + assertThat(resultTopic).isTrue(); + + boolean resultGroup = BrokerMetricsManager.isSystem(topic, systemGroup); + assertThat(resultGroup).isTrue(); + } + + @Test + public void testIsSystem_NonSystemTopicAndGroup_ReturnsFalse() { + String topic = "FooTopic"; + String group = "FooGroup"; + boolean result = BrokerMetricsManager.isSystem(topic, group); + assertThat(result).isFalse(); + } + + @Test + public void testIsSystem_EmptyTopicAndGroup_ReturnsFalse() { + String topic = ""; + String group = ""; + boolean result = BrokerMetricsManager.isSystem(topic, group); + assertThat(result).isFalse(); + } + + @Test + public void testGetMessageTypeAsNormal() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + requestHeader.setProperties(""); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.NORMAL).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsTransaction() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TRANSACTION_PREPARED, "true"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.TRANSACTION).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsFifo() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_SHARDING_KEY, "shardingKey"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.FIFO).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsDelayLevel() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, "1"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.DELAY).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsDeliverMS() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TIMER_DELIVER_MS, "10"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.DELAY).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsDelaySEC() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TIMER_DELAY_SEC, "1"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.DELAY).isEqualTo(result); + } + + @Test + public void testGetMessageTypeAsDelayMS() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TIMER_DELAY_MS, "10"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.DELAY).isEqualTo(result); + } + + @Test + public void testGetMessageTypeWithUnknownProperty() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put("unknownProperty", "unknownValue"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.NORMAL).isEqualTo(result); + } + + @Test + public void testGetMessageTypeWithMultipleProperties() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, "1"); + map.put(MessageConst.PROPERTY_SHARDING_KEY, "shardingKey"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.FIFO).isEqualTo(result); + } + + @Test + public void testGetMessageTypeWithTransactionFlagButOtherPropertiesPresent() { + SendMessageRequestHeader requestHeader = new SendMessageRequestHeader(); + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TRANSACTION_PREPARED, "true"); + map.put(MessageConst.PROPERTY_SHARDING_KEY, "shardingKey"); + requestHeader.setProperties(MessageDecoder.messageProperties2String(map)); + + TopicMessageType result = BrokerMetricsManager.getMessageType(requestHeader); + assertThat(TopicMessageType.TRANSACTION).isEqualTo(result); + } + + @Test + public void testGetMessageTypeWithEmptyProperties() { + TopicMessageType result = BrokerMetricsManager.getMessageType(new SendMessageRequestHeader()); + assertThat(TopicMessageType.NORMAL).isEqualTo(result); + } + + @Test + public void testCreateMetricsManager() { + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + String storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator + "store-" + + UUID.randomUUID(); + messageStoreConfig.setStorePathRootDir(storePathRootDir); + BrokerConfig brokerConfig = new BrokerConfig(); + + NettyServerConfig nettyServerConfig = new NettyServerConfig(); + nettyServerConfig.setListenPort(0); + + BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, + new NettyClientConfig(), messageStoreConfig); + + BrokerMetricsManager metricsManager = new BrokerMetricsManager(brokerController); + + assertThat(metricsManager.getBrokerMeter()).isNull(); + } + + @Test + public void testCreateMetricsManagerLogType() throws CloneNotSupportedException { + BrokerConfig brokerConfig = new BrokerConfig(); + brokerConfig.setMetricsExporterType(MetricsExporterType.LOG); + brokerConfig.setMetricsLabel("label1:value1;label2:value2"); + brokerConfig.setMetricsOtelCardinalityLimit(1); + + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + String storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator + "store-" + + UUID.randomUUID(); + messageStoreConfig.setStorePathRootDir(storePathRootDir); + + NettyServerConfig nettyServerConfig = new NettyServerConfig(); + nettyServerConfig.setListenPort(0); + + BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, + new NettyClientConfig(), messageStoreConfig); + brokerController.initialize(); + + BrokerMetricsManager metricsManager = new BrokerMetricsManager(brokerController); + + assertThat(metricsManager.getBrokerMeter()).isNotNull(); + } } \ No newline at end of file diff --git a/broker/src/test/java/org/apache/rocketmq/broker/offset/RocksDBLmqConsumerOffsetManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/offset/RocksDBLmqConsumerOffsetManagerTest.java new file mode 100644 index 00000000000..ea6528546dc --- /dev/null +++ b/broker/src/test/java/org/apache/rocketmq/broker/offset/RocksDBLmqConsumerOffsetManagerTest.java @@ -0,0 +1,134 @@ +/* + * 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.broker.offset; + +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.when; + +public class RocksDBLmqConsumerOffsetManagerTest { + private static final String LMQ_GROUP = MixAll.LMQ_PREFIX + "FooBarGroup"; + private static final String NON_LMQ_GROUP = "nonLmqGroup"; + private static final String TOPIC = "FooBarTopic"; + private static final int QUEUE_ID = 0; + private static final long OFFSET = 12345; + + private BrokerController brokerController; + + private RocksDBLmqConsumerOffsetManager offsetManager; + + @Before + public void setUp() { + brokerController = Mockito.mock(BrokerController.class); + when(brokerController.getMessageStoreConfig()).thenReturn(Mockito.mock(MessageStoreConfig.class)); + when(brokerController.getBrokerConfig()).thenReturn(Mockito.mock(BrokerConfig.class)); + offsetManager = new RocksDBLmqConsumerOffsetManager(brokerController); + } + + @Test + public void testQueryOffsetForLmq() { + // Setup + offsetManager.getLmqOffsetTable().put(getKey(), OFFSET); + // Execute + long actualOffset = offsetManager.queryOffset(LMQ_GROUP, TOPIC, QUEUE_ID); + // Verify + assertEquals("Offset should match the expected value.", OFFSET, actualOffset); + } + + @Test + public void testQueryOffsetForNonLmq() { + long actualOffset = offsetManager.queryOffset(NON_LMQ_GROUP, TOPIC, QUEUE_ID); + // Verify + assertEquals("Offset should not be null.", -1, actualOffset); + } + + + @Test + public void testQueryOffsetForLmqGroupWithExistingOffset() { + offsetManager.getLmqOffsetTable().put(getKey(), OFFSET); + + // Act + Map actualOffsets = offsetManager.queryOffset(LMQ_GROUP, TOPIC); + + // Assert + assertNotNull(actualOffsets); + assertEquals(1, actualOffsets.size()); + assertEquals(OFFSET, (long) actualOffsets.get(0)); + } + + @Test + public void testQueryOffsetForLmqGroupWithoutExistingOffset() { + // Act + Map actualOffsets = offsetManager.queryOffset(LMQ_GROUP, "nonExistingTopic"); + + // Assert + assertNotNull(actualOffsets); + assertTrue("The map should be empty for non-existing offsets", actualOffsets.isEmpty()); + } + + @Test + public void testQueryOffsetForNonLmqGroup() { + when(brokerController.getBrokerConfig().getConsumerOffsetUpdateVersionStep()).thenReturn(1L); + // Arrange + Map mockOffsets = new HashMap<>(); + mockOffsets.put(QUEUE_ID, OFFSET); + + offsetManager.commitOffset("clientHost", NON_LMQ_GROUP, TOPIC, QUEUE_ID, OFFSET); + + // Act + Map actualOffsets = offsetManager.queryOffset(NON_LMQ_GROUP, TOPIC); + + // Assert + assertNotNull(actualOffsets); + assertEquals("Offsets should match the mocked return value for non-LMQ groups", mockOffsets, actualOffsets); + } + + @Test + public void testCommitOffsetForLmq() { + // Execute + offsetManager.commitOffset("clientHost", LMQ_GROUP, TOPIC, QUEUE_ID, OFFSET); + // Verify + Long expectedOffset = offsetManager.getLmqOffsetTable().get(getKey()); + assertEquals("Offset should be updated correctly.", OFFSET, expectedOffset.longValue()); + } + + @Test + public void testEncode() { + offsetManager.setLmqOffsetTable(new ConcurrentHashMap<>(512)); + offsetManager.getLmqOffsetTable().put(getKey(), OFFSET); + String encodedData = offsetManager.encode(); + assertTrue(encodedData.contains(String.valueOf(OFFSET))); + } + + private String getKey() { + return TOPIC + "@" + LMQ_GROUP; + } +} diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PeekMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PeekMessageProcessorTest.java new file mode 100644 index 00000000000..7f8504453ca --- /dev/null +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PeekMessageProcessorTest.java @@ -0,0 +1,170 @@ +/* + * 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.broker.processor; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; +import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; +import org.apache.rocketmq.broker.topic.TopicConfigManager; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.TopicConfig; +import org.apache.rocketmq.common.constant.PermName; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.netty.NettyClientConfig; +import org.apache.rocketmq.remoting.netty.NettyServerConfig; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.remoting.protocol.RequestCode; +import org.apache.rocketmq.remoting.protocol.ResponseCode; +import org.apache.rocketmq.remoting.protocol.header.PeekMessageRequestHeader; +import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; +import org.apache.rocketmq.store.GetMessageResult; +import org.apache.rocketmq.store.GetMessageStatus; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.SelectMappedBufferResult; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; + +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; + +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class PeekMessageProcessorTest { + + private PeekMessageProcessor peekMessageProcessor; + + @Spy + private BrokerController brokerController = new BrokerController(new BrokerConfig(), new NettyServerConfig(), new NettyClientConfig(), new MessageStoreConfig()); + + @Mock + private ChannelHandlerContext handlerContext; + + @Mock + private MessageStore messageStore; + + @Mock + private SubscriptionGroupManager subscriptionGroupManager; + + @Mock + private ConsumerOffsetManager consumerOffsetManager; + + @Mock + private SubscriptionGroupConfig subscriptionGroupConfig; + + @Mock + private Channel channel; + + private TopicConfigManager topicConfigManager; + + @Before + public void init() { + peekMessageProcessor = new PeekMessageProcessor(brokerController); + when(brokerController.getMessageStore()).thenReturn(messageStore); + topicConfigManager = new TopicConfigManager(brokerController); + when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager); + when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); + when(subscriptionGroupManager.findSubscriptionGroupConfig(anyString())).thenReturn(subscriptionGroupConfig); + when(subscriptionGroupConfig.isConsumeEnable()).thenReturn(true); + topicConfigManager.getTopicConfigTable().put("topic", new TopicConfig("topic")); + when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager); + when(consumerOffsetManager.queryOffset(anyString(), anyString(), anyInt())).thenReturn(-1L); + when(messageStore.getMinOffsetInQueue(anyString(),anyInt())).thenReturn(0L); + when(handlerContext.channel()).thenReturn(channel); + when(channel.remoteAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 12345)); + } + + @Test + public void testProcessRequest() throws RemotingCommandException { + RemotingCommand request = createPeekMessageRequest("group","topic",0); + GetMessageResult getMessageResult = new GetMessageResult(); + getMessageResult.setStatus(GetMessageStatus.FOUND); + ByteBuffer bb = ByteBuffer.allocate(64); + bb.putLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSITION, System.currentTimeMillis()); + SelectMappedBufferResult mappedBufferResult1 = new SelectMappedBufferResult(0, bb, 64, null); + for (int i = 0; i < 10;i++) { + getMessageResult.addMessage(mappedBufferResult1); + } + when(messageStore.getMessage(anyString(),anyString(),anyInt(),anyLong(),anyInt(),any())).thenReturn(getMessageResult); + RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS); + } + + @Test + public void testProcessRequest_NoPermission() throws RemotingCommandException { + this.brokerController.getBrokerConfig().setBrokerPermission(PermName.PERM_WRITE); + RemotingCommand request = createPeekMessageRequest("group","topic",0); + RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION); + this.brokerController.getBrokerConfig().setBrokerPermission(PermName.PERM_WRITE | PermName.PERM_READ); + + topicConfigManager.getTopicConfigTable().get("topic").setPerm(PermName.PERM_WRITE); + response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION); + topicConfigManager.getTopicConfigTable().get("topic").setPerm(PermName.PERM_WRITE | PermName.PERM_READ); + + when(subscriptionGroupConfig.isConsumeEnable()).thenReturn(false); + response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION); + } + + @Test + public void testProcessRequest_TopicNotExist() throws RemotingCommandException { + RemotingCommand request = createPeekMessageRequest("group1","topic1",0); + RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.TOPIC_NOT_EXIST); + } + + @Test + public void testProcessRequest_SubscriptionGroupNotExist() throws RemotingCommandException { + when(subscriptionGroupManager.findSubscriptionGroupConfig(anyString())).thenReturn(null); + RemotingCommand request = createPeekMessageRequest("group","topic",0); + RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST); + } + + @Test + public void testProcessRequest_QueueIdError() throws RemotingCommandException { + RemotingCommand request = createPeekMessageRequest("group","topic",17); + RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request); + assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR); + } + + private RemotingCommand createPeekMessageRequest(String group,String topic,int queueId) { + PeekMessageRequestHeader peekMessageRequestHeader = new PeekMessageRequestHeader(); + peekMessageRequestHeader.setConsumerGroup(group); + peekMessageRequestHeader.setTopic(topic); + peekMessageRequestHeader.setMaxMsgNums(10); + peekMessageRequestHeader.setQueueId(queueId); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PEEK_MESSAGE, peekMessageRequestHeader); + request.makeCustomHeaderToNet(); + return request; + } +} diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyServiceTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyServiceTest.java new file mode 100644 index 00000000000..257783ecb48 --- /dev/null +++ b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyServiceTest.java @@ -0,0 +1,243 @@ +/* + * 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.impl.consumer; + +import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +import org.apache.rocketmq.client.impl.factory.MQClientInstance; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.stat.ConsumerStatsManager; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.remoting.protocol.body.CMResult; +import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadPoolExecutor; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class ConsumeMessagePopOrderlyServiceTest { + + @Mock + private DefaultMQPushConsumerImpl defaultMQPushConsumerImpl; + + @Mock + private MessageListenerOrderly messageListener; + + @Mock + private DefaultMQPushConsumer defaultMQPushConsumer; + + @Mock + private ConsumerStatsManager consumerStatsManager; + + @Mock + private RebalanceImpl rebalanceImpl; + + private ConsumeMessagePopOrderlyService popService; + + private final String defaultGroup = "defaultGroup"; + + private final String defaultBroker = "defaultBroker"; + + private final String defaultTopic = "defaultTopic"; + + @Before + public void init() throws Exception { + when(defaultMQPushConsumer.getConsumerGroup()).thenReturn(defaultGroup); + when(defaultMQPushConsumer.getConsumeThreadMin()).thenReturn(1); + when(defaultMQPushConsumer.getConsumeThreadMax()).thenReturn(3); + when(defaultMQPushConsumerImpl.getDefaultMQPushConsumer()).thenReturn(defaultMQPushConsumer); + when(defaultMQPushConsumerImpl.getRebalanceImpl()).thenReturn(rebalanceImpl); + when(defaultMQPushConsumerImpl.getConsumerStatsManager()).thenReturn(consumerStatsManager); + MQClientInstance mQClientFactory = mock(MQClientInstance.class); + DefaultMQProducer defaultMQProducer = mock(DefaultMQProducer.class); + when(mQClientFactory.getDefaultMQProducer()).thenReturn(defaultMQProducer); + when(defaultMQPushConsumerImpl.getmQClientFactory()).thenReturn(mQClientFactory); + popService = new ConsumeMessagePopOrderlyService(defaultMQPushConsumerImpl, messageListener); + } + + @Test + public void testShutdown() throws IllegalAccessException { + popService.shutdown(3000L); + Field scheduledExecutorServiceField = FieldUtils.getDeclaredField(popService.getClass(), "scheduledExecutorService", true); + Field consumeExecutorField = FieldUtils.getDeclaredField(popService.getClass(), "consumeExecutor", true); + ScheduledExecutorService scheduledExecutorService = (ScheduledExecutorService) scheduledExecutorServiceField.get(popService); + ThreadPoolExecutor consumeExecutor = (ThreadPoolExecutor) consumeExecutorField.get(popService); + assertTrue(scheduledExecutorService.isShutdown()); + assertTrue(scheduledExecutorService.isTerminated()); + assertTrue(consumeExecutor.isShutdown()); + assertTrue(consumeExecutor.isTerminated()); + } + + @Test + public void testUnlockAllMessageQueues() { + popService.unlockAllMessageQueues(); + verify(rebalanceImpl, times(1)).unlockAll(eq(false)); + } + + @Test + public void testUpdateCorePoolSize() { + popService.updateCorePoolSize(2); + popService.incCorePoolSize(); + popService.decCorePoolSize(); + assertEquals(2, popService.getCorePoolSize()); + } + + @Test + public void testConsumeMessageDirectly() { + when(messageListener.consumeMessage(any(), any(ConsumeOrderlyContext.class))).thenReturn(ConsumeOrderlyStatus.SUCCESS); + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_SUCCESS, actual.getConsumeResult()); + assertTrue(actual.isOrder()); + } + + @Test + public void testConsumeMessageDirectlyWithCommit() { + when(messageListener.consumeMessage(any(), any(ConsumeOrderlyContext.class))).thenReturn(ConsumeOrderlyStatus.COMMIT); + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_COMMIT, actual.getConsumeResult()); + assertTrue(actual.isOrder()); + } + + @Test + public void testConsumeMessageDirectlyWithRollback() { + when(messageListener.consumeMessage(any(), any(ConsumeOrderlyContext.class))).thenReturn(ConsumeOrderlyStatus.ROLLBACK); + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_ROLLBACK, actual.getConsumeResult()); + assertTrue(actual.isOrder()); + } + + @Test + public void testConsumeMessageDirectlyWithCrLater() { + when(messageListener.consumeMessage(any(), any(ConsumeOrderlyContext.class))).thenReturn(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT); + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_LATER, actual.getConsumeResult()); + } + + @Test + public void testConsumeMessageDirectlyWithCrReturnNull() { + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_RETURN_NULL, actual.getConsumeResult()); + } + + @Test + public void testConsumeMessageDirectlyWithCrThrowException() { + when(messageListener.consumeMessage(any(), any(ConsumeOrderlyContext.class))).thenThrow(new RuntimeException("exception")); + ConsumeMessageDirectlyResult actual = popService.consumeMessageDirectly(createMessageExt(), defaultBroker); + assertEquals(CMResult.CR_THROW_EXCEPTION, actual.getConsumeResult()); + } + + @Test + public void testSubmitConsumeRequest() { + assertThrows(UnsupportedOperationException.class, () -> { + List msgs = mock(List.class); + ProcessQueue processQueue = mock(ProcessQueue.class); + MessageQueue messageQueue = mock(MessageQueue.class); + popService.submitConsumeRequest(msgs, processQueue, messageQueue, false); + }); + } + + @Test + public void testSubmitPopConsumeRequest() throws IllegalAccessException { + List msgs = Collections.singletonList(createMessageExt()); + PopProcessQueue processQueue = mock(PopProcessQueue.class); + MessageQueue messageQueue = mock(MessageQueue.class); + ThreadPoolExecutor consumeExecutor = mock(ThreadPoolExecutor.class); + FieldUtils.writeDeclaredField(popService, "consumeExecutor", consumeExecutor, true); + popService.submitPopConsumeRequest(msgs, processQueue, messageQueue); + verify(consumeExecutor, times(1)).submit(any(Runnable.class)); + } + + @Test + public void testLockMQPeriodically() { + popService.lockMQPeriodically(); + verify(defaultMQPushConsumerImpl, times(1)).getRebalanceImpl(); + verify(rebalanceImpl, times(1)).lockAll(); + } + + @Test + public void testGetConsumerStatsManager() { + ConsumerStatsManager actual = popService.getConsumerStatsManager(); + assertNotNull(actual); + assertEquals(consumerStatsManager, actual); + } + + @Test + public void testSendMessageBack() { + assertTrue(popService.sendMessageBack(createMessageExt())); + } + + @Test + public void testProcessConsumeResult() { + ConsumeOrderlyContext context = mock(ConsumeOrderlyContext.class); + ConsumeMessagePopOrderlyService.ConsumeRequest consumeRequest = mock(ConsumeMessagePopOrderlyService.ConsumeRequest.class); + assertTrue(popService.processConsumeResult(Collections.singletonList(createMessageExt()), ConsumeOrderlyStatus.SUCCESS, context, consumeRequest)); + } + + @Test + public void testResetNamespace() { + when(defaultMQPushConsumer.getNamespace()).thenReturn("defaultNamespace"); + List msgs = Collections.singletonList(createMessageExt()); + popService.resetNamespace(msgs); + assertEquals(defaultTopic, msgs.get(0).getTopic()); + } + + private MessageExt createMessageExt() { + MessageExt result = new MessageExt(); + result.setBody("body".getBytes(StandardCharsets.UTF_8)); + result.setTopic(defaultTopic); + result.setBrokerName(defaultBroker); + result.putUserProperty("key", "value"); + result.getProperties().put(MessageConst.PROPERTY_PRODUCER_GROUP, defaultGroup); + result.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, "TX1"); + long curTime = System.currentTimeMillis(); + result.setBornTimestamp(curTime - 1000); + result.getProperties().put(MessageConst.PROPERTY_POP_CK, curTime + " " + curTime + " " + curTime + " " + curTime); + result.setKeys("keys"); + SocketAddress bornHost = new InetSocketAddress("127.0.0.1", 12911); + SocketAddress storeHost = new InetSocketAddress("127.0.0.1", 10911); + result.setBornHost(bornHost); + result.setStoreHost(storeHost); + return result; + } +} diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapperTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapperTest.java new file mode 100644 index 00000000000..2ffa8f4f149 --- /dev/null +++ b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapperTest.java @@ -0,0 +1,244 @@ +/* + * 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.impl.consumer; + +import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.rocketmq.client.ClientConfig; +import org.apache.rocketmq.client.consumer.PopCallback; +import org.apache.rocketmq.client.consumer.PullCallback; +import org.apache.rocketmq.client.consumer.PullResult; +import org.apache.rocketmq.client.consumer.PullStatus; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.hook.FilterMessageContext; +import org.apache.rocketmq.client.hook.FilterMessageHook; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.impl.FindBrokerResult; +import org.apache.rocketmq.client.impl.MQClientAPIImpl; +import org.apache.rocketmq.client.impl.factory.MQClientInstance; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.sysflag.MessageSysFlag; +import org.apache.rocketmq.common.sysflag.PullSysFlag; +import org.apache.rocketmq.remoting.exception.RemotingException; +import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader; +import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData; +import org.apache.rocketmq.remoting.protocol.route.BrokerData; +import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class PullAPIWrapperTest { + + @Mock + private MQClientInstance mQClientFactory; + + @Mock + private MQClientAPIImpl mqClientAPIImpl; + + private PullAPIWrapper pullAPIWrapper; + + private final String defaultGroup = "defaultGroup"; + + private final String defaultBroker = "defaultBroker"; + + private final String defaultTopic = "defaultTopic"; + + private final String defaultBrokerAddr = "127.0.0.1:10911"; + + private final long defaultTimeout = 3000L; + + @Before + public void init() throws Exception { + ClientConfig clientConfig = mock(ClientConfig.class); + when(mQClientFactory.getClientConfig()).thenReturn(clientConfig); + MQClientAPIImpl mqClientAPIImpl = mock(MQClientAPIImpl.class); + when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPIImpl); + when(mQClientFactory.getTopicRouteTable()).thenReturn(createTopicRouteTable()); + FindBrokerResult findBrokerResult = mock(FindBrokerResult.class); + when(findBrokerResult.getBrokerAddr()).thenReturn(defaultBrokerAddr); + when(mQClientFactory.findBrokerAddressInSubscribe(any(), anyLong(), anyBoolean())).thenReturn(findBrokerResult); + pullAPIWrapper = new PullAPIWrapper(mQClientFactory, defaultGroup, false); + ArrayList filterMessageHookList = new ArrayList<>(); + filterMessageHookList.add(mock(FilterMessageHook.class)); + FieldUtils.writeDeclaredField(pullAPIWrapper, "filterMessageHookList", filterMessageHookList, true); + } + + @Test + public void testProcessPullResult() throws Exception { + PullResultExt pullResult = mock(PullResultExt.class); + when(pullResult.getPullStatus()).thenReturn(PullStatus.FOUND); + when(pullResult.getMessageBinary()).thenReturn(MessageDecoder.encode(createMessageExt(), false)); + SubscriptionData subscriptionData = mock(SubscriptionData.class); + PullResult actual = pullAPIWrapper.processPullResult(createMessageQueue(), pullResult, subscriptionData); + assertNotNull(actual); + assertEquals(0, actual.getNextBeginOffset()); + assertEquals(0, actual.getMsgFoundList().size()); + } + + @Test + public void testExecuteHook() throws IllegalAccessException { + FilterMessageContext filterMessageContext = mock(FilterMessageContext.class); + ArrayList filterMessageHookList = new ArrayList<>(); + FilterMessageHook filterMessageHook = mock(FilterMessageHook.class); + filterMessageHookList.add(filterMessageHook); + FieldUtils.writeDeclaredField(pullAPIWrapper, "filterMessageHookList", filterMessageHookList, true); + pullAPIWrapper.executeHook(filterMessageContext); + verify(filterMessageHook, times(1)).filterMessage(any(FilterMessageContext.class)); + } + + @Test + public void testPullKernelImpl() throws Exception { + PullCallback pullCallback = mock(PullCallback.class); + when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPIImpl); + PullResult actual = pullAPIWrapper.pullKernelImpl(createMessageQueue(), + "", + "", + 1L, + 1L, + 1, + 1, + PullSysFlag.buildSysFlag(false, false, false, true), + 1L, + System.currentTimeMillis(), + defaultTimeout, CommunicationMode.ASYNC, pullCallback); + assertNull(actual); + verify(mqClientAPIImpl, times(1)).pullMessage(eq(defaultBroker), + any(PullMessageRequestHeader.class), + eq(defaultTimeout), + any(CommunicationMode.class), + any(PullCallback.class)); + } + + @Test + public void testSetConnectBrokerByUser() { + pullAPIWrapper.setConnectBrokerByUser(true); + assertTrue(pullAPIWrapper.isConnectBrokerByUser()); + } + + @Test + public void testRandomNum() { + int randomNum = pullAPIWrapper.randomNum(); + assertTrue(randomNum > 0); + } + + @Test + public void testSetDefaultBrokerId() { + pullAPIWrapper.setDefaultBrokerId(MixAll.MASTER_ID); + assertEquals(MixAll.MASTER_ID, pullAPIWrapper.getDefaultBrokerId()); + } + + @Test + public void testPopAsync() throws RemotingException, InterruptedException, MQClientException { + PopCallback popCallback = mock(PopCallback.class); + when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPIImpl); + pullAPIWrapper.popAsync(createMessageQueue(), + System.currentTimeMillis(), + 1, + defaultGroup, + defaultTimeout, + popCallback, + true, + 1, + false, + "", + ""); + verify(mqClientAPIImpl, times(1)).popMessageAsync(eq(defaultBroker), + eq(defaultBrokerAddr), + any(PopMessageRequestHeader.class), + eq(13000L), + any(PopCallback.class)); + } + + private ConcurrentMap createTopicRouteTable() { + TopicRouteData topicRouteData = new TopicRouteData(); + List brokerDatas = new ArrayList<>(); + BrokerData brokerData = new BrokerData(); + brokerData.setBrokerName(defaultBroker); + HashMap brokerAddrs = new HashMap<>(); + brokerAddrs.put(MixAll.MASTER_ID, defaultBroker); + brokerData.setBrokerAddrs(brokerAddrs); + brokerDatas.add(brokerData); + topicRouteData.setBrokerDatas(brokerDatas); + HashMap> filterServerTable = new HashMap<>(); + List filterServers = new ArrayList<>(); + filterServers.add(defaultBroker); + filterServerTable.put(defaultBrokerAddr, filterServers); + topicRouteData.setFilterServerTable(filterServerTable); + ConcurrentMap result = new ConcurrentHashMap<>(); + result.put(defaultTopic, topicRouteData); + return result; + } + + private MessageQueue createMessageQueue() { + MessageQueue result = new MessageQueue(); + result.setQueueId(0); + result.setBrokerName(defaultBroker); + result.setTopic(defaultTopic); + return result; + } + + private MessageExt createMessageExt() { + MessageExt result = new MessageExt(); + result.setBody("body".getBytes(StandardCharsets.UTF_8)); + result.setTopic(defaultTopic); + result.setBrokerName(defaultBroker); + result.putUserProperty("key", "value"); + result.getProperties().put(MessageConst.PROPERTY_PRODUCER_GROUP, defaultGroup); + result.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, "TX1"); + long curTime = System.currentTimeMillis(); + result.setBornTimestamp(curTime - 1000); + result.getProperties().put(MessageConst.PROPERTY_POP_CK, curTime + " " + curTime + " " + curTime + " " + curTime); + result.setKeys("keys"); + result.setSysFlag(MessageSysFlag.INNER_BATCH_FLAG); + result.setSysFlag(result.getSysFlag() | MessageSysFlag.NEED_UNWRAP_FLAG); + SocketAddress bornHost = new InetSocketAddress("127.0.0.1", 12911); + SocketAddress storeHost = new InetSocketAddress("127.0.0.1", 10911); + result.setBornHost(bornHost); + result.setStoreHost(storeHost); + return result; + } +} diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/ProduceAccumulatorTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/ProduceAccumulatorTest.java index 7074fae243d..8e76238d47f 100644 --- a/client/src/test/java/org/apache/rocketmq/client/producer/ProduceAccumulatorTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/producer/ProduceAccumulatorTest.java @@ -106,6 +106,7 @@ public void testProduceAccumulator_sync() throws MQBrokerException, RemotingExce final MockMQProducer mockMQProducer = new MockMQProducer(); final ProduceAccumulator produceAccumulator = new ProduceAccumulator("test"); + produceAccumulator.batchMaxDelayMs(3000); produceAccumulator.start(); List messages = new ArrayList(); @@ -134,7 +135,7 @@ public void run() { } }).start(); } - assertThat(countDownLatch.await(3000L, TimeUnit.MILLISECONDS)).isTrue(); + assertThat(countDownLatch.await(5000L, TimeUnit.MILLISECONDS)).isTrue(); assertThat(mockMQProducer.beSendMessage instanceof MessageBatch).isTrue(); MessageBatch messageBatch1 = (MessageBatch) mockMQProducer.beSendMessage; diff --git a/common/src/test/java/org/apache/rocketmq/common/chain/HandlerChainTest.java b/common/src/test/java/org/apache/rocketmq/common/chain/HandlerChainTest.java new file mode 100644 index 00000000000..3a8499ebad2 --- /dev/null +++ b/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 handlerChain; + private Handler handler1; + private Handler 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); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/coldctr/AccAndTimeStampTest.java b/common/src/test/java/org/apache/rocketmq/common/coldctr/AccAndTimeStampTest.java new file mode 100644 index 00000000000..01bb4ae3701 --- /dev/null +++ b/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())); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTypeTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTypeTest.java new file mode 100644 index 00000000000..f9586bd2da8 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/compression/CompressionTypeTest.java @@ -0,0 +1,57 @@ +/* + * 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.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +public class CompressionTypeTest { + + @Test + public void testCompressionTypeValues() { + assertEquals(1, CompressionType.LZ4.getValue()); + assertEquals(2, CompressionType.ZSTD.getValue()); + assertEquals(3, CompressionType.ZLIB.getValue()); + } + + @Test + public void testCompressionTypeOf() { + assertEquals(CompressionType.LZ4, CompressionType.of("LZ4")); + assertEquals(CompressionType.ZSTD, CompressionType.of("ZSTD")); + assertEquals(CompressionType.ZLIB, CompressionType.of("ZLIB")); + assertThrows(RuntimeException.class, () -> CompressionType.of("UNKNOWN")); + } + + @Test + public void testCompressionTypeFindByValue() { + assertEquals(CompressionType.LZ4, CompressionType.findByValue(1)); + assertEquals(CompressionType.ZSTD, CompressionType.findByValue(2)); + assertEquals(CompressionType.ZLIB, CompressionType.findByValue(3)); + assertEquals(CompressionType.ZLIB, CompressionType.findByValue(0)); + assertThrows(RuntimeException.class, () -> CompressionType.findByValue(99)); + } + + @Test + public void testCompressionFlag() { + assertEquals(MessageSysFlag.COMPRESSION_LZ4_TYPE, CompressionType.LZ4.getCompressionFlag()); + assertEquals(MessageSysFlag.COMPRESSION_ZSTD_TYPE, CompressionType.ZSTD.getCompressionFlag()); + assertEquals(MessageSysFlag.COMPRESSION_ZLIB_TYPE, CompressionType.ZLIB.getCompressionFlag()); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/CompressorFactoryTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/CompressorFactoryTest.java new file mode 100644 index 00000000000..e150fb2f7aa --- /dev/null +++ b/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); + } + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/Lz4CompressorTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/Lz4CompressorTest.java new file mode 100644 index 00000000000..ca59025c133 --- /dev/null +++ b/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); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/ZlibCompressorTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/ZlibCompressorTest.java new file mode 100644 index 00000000000..f46ac7c6691 --- /dev/null +++ b/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 + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/compression/ZstdCompressorTest.java b/common/src/test/java/org/apache/rocketmq/common/compression/ZstdCompressorTest.java new file mode 100644 index 00000000000..574e1281811 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/compression/ZstdCompressorTest.java @@ -0,0 +1,78 @@ +/* + * 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 org.junit.Test; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertTrue; + +public class ZstdCompressorTest { + + @Test + public void testCompressAndDecompress() throws IOException { + byte[] originalData = "RocketMQ is awesome!".getBytes(); + ZstdCompressor compressor = new ZstdCompressor(); + 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 testCompressWithInvalidData() throws IOException { + byte[] invalidData = new byte[] {-1, -1, -1, -1}; + ZstdCompressor compressor = new ZstdCompressor(); + compressor.compress(invalidData, 1); + } + + @Test(expected = IOException.class) + public void testDecompressWithInvalidData() throws IOException { + byte[] invalidData = new byte[] {-1, -1, -1, -1}; + ZstdCompressor compressor = new ZstdCompressor(); + compressor.decompress(invalidData); + } + + @Test + public void testCompressAndDecompressEmptyString() throws IOException { + byte[] originalData = "".getBytes(); + ZstdCompressor compressor = new ZstdCompressor(); + byte[] compressedData = compressor.compress(originalData, 1); + assertTrue("Compressed data for empty string should not be empty", compressedData.length > 0); + + byte[] decompressedData = compressor.decompress(compressedData); + assertArrayEquals("Decompressed data for empty string should match original", originalData, decompressedData); + } + + @Test + public void testCompressAndDecompressLargeData() throws IOException { + StringBuilder largeStringBuilder = new StringBuilder(); + for (int i = 0; i < 10000; i++) { + largeStringBuilder.append("RocketMQ is awesome! "); + } + byte[] originalData = largeStringBuilder.toString().getBytes(); + + ZstdCompressor compressor = new ZstdCompressor(); + byte[] compressedData = compressor.compress(originalData, 1); + assertTrue("Compressed data for large data should be smaller than original", compressedData.length < originalData.length); + + byte[] decompressedData = compressor.decompress(compressedData); + assertArrayEquals("Decompressed data for large data should match original", originalData, decompressedData); + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/consumer/ReceiptHandleTest.java b/common/src/test/java/org/apache/rocketmq/common/consumer/ReceiptHandleTest.java new file mode 100644 index 00000000000..54741817e12 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/consumer/ReceiptHandleTest.java @@ -0,0 +1,103 @@ +/* + * 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.consumer; + +import org.apache.rocketmq.common.KeyBuilder; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class ReceiptHandleTest { + + @Test + public void testEncodeAndDecode() { + long startOffset = 1000L; + long retrieveTime = System.currentTimeMillis(); + long invisibleTime = 1000L; + int reviveQueueId = 1; + String topicType = "NORMAL"; + String brokerName = "BrokerA"; + int queueId = 2; + long offset = 2000L; + long commitLogOffset = 3000L; + ReceiptHandle receiptHandle = ReceiptHandle.builder() + .startOffset(startOffset) + .retrieveTime(retrieveTime) + .invisibleTime(invisibleTime) + .reviveQueueId(reviveQueueId) + .topicType(topicType) + .brokerName(brokerName) + .queueId(queueId) + .offset(offset) + .commitLogOffset(commitLogOffset) + .build(); + + String encoded = receiptHandle.encode(); + ReceiptHandle decoded = ReceiptHandle.decode(encoded); + + assertEquals(receiptHandle.getStartOffset(), decoded.getStartOffset()); + assertEquals(receiptHandle.getRetrieveTime(), decoded.getRetrieveTime()); + assertEquals(receiptHandle.getInvisibleTime(), decoded.getInvisibleTime()); + assertEquals(receiptHandle.getReviveQueueId(), decoded.getReviveQueueId()); + assertEquals(receiptHandle.getTopicType(), decoded.getTopicType()); + assertEquals(receiptHandle.getBrokerName(), decoded.getBrokerName()); + assertEquals(receiptHandle.getQueueId(), decoded.getQueueId()); + assertEquals(receiptHandle.getOffset(), decoded.getOffset()); + assertEquals(receiptHandle.getCommitLogOffset(), decoded.getCommitLogOffset()); + } + + @Test(expected = IllegalArgumentException.class) + public void testDecodeWithInvalidString() { + String invalidReceiptHandle = "invalid_data"; + + ReceiptHandle.decode(invalidReceiptHandle); + } + + @Test + public void testIsExpired() { + long startOffset = 1000L; + long retrieveTime = System.currentTimeMillis(); + long invisibleTime = 1000L; + int reviveQueueId = 1; + String topicType = "NORMAL"; + String brokerName = "BrokerA"; + int queueId = 2; + long offset = 2000L; + long commitLogOffset = 3000L; + long pastTime = System.currentTimeMillis() - 1000L; + ReceiptHandle receiptHandle = new ReceiptHandle(startOffset, retrieveTime, invisibleTime, pastTime, reviveQueueId, topicType, brokerName, queueId, offset, commitLogOffset, ""); + + boolean isExpired = receiptHandle.isExpired(); + + assertTrue(isExpired); + } + + @Test + public void testGetRealTopic() { + // Arrange + String topic = "TestTopic"; + String groupName = "TestGroup"; + ReceiptHandle receiptHandle = ReceiptHandle.builder() + .topicType(ReceiptHandle.RETRY_TOPIC) + .build(); + + String realTopic = receiptHandle.getRealTopic(topic, groupName); + + assertEquals(KeyBuilder.buildPopRetryTopicV1(topic, groupName), realTopic); + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java index c671593a34b..05d8e5fbe13 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java @@ -125,22 +125,30 @@ protected RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCo final String producerGroup = requestHeader.getProducerGroup(); if (producerGroup != null) { RemotingChannel channel = this.remotingChannelManager.removeProducerChannel(context, producerGroup, ctx.channel()); - ClientChannelInfo clientChannelInfo = new ClientChannelInfo( - channel, - requestHeader.getClientID(), - request.getLanguage(), - request.getVersion()); - this.messagingProcessor.unRegisterProducer(context, producerGroup, clientChannelInfo); + if (channel != null) { + ClientChannelInfo clientChannelInfo = new ClientChannelInfo( + channel, + requestHeader.getClientID(), + request.getLanguage(), + request.getVersion()); + this.messagingProcessor.unRegisterProducer(context, producerGroup, clientChannelInfo); + } else { + log.warn("unregister producer failed, channel not exist, may has been removed, producerGroup={}, channel={}", producerGroup, ctx.channel()); + } } final String consumerGroup = requestHeader.getConsumerGroup(); if (consumerGroup != null) { RemotingChannel channel = this.remotingChannelManager.removeConsumerChannel(context, consumerGroup, ctx.channel()); - ClientChannelInfo clientChannelInfo = new ClientChannelInfo( - channel, - requestHeader.getClientID(), - request.getLanguage(), - request.getVersion()); - this.messagingProcessor.unRegisterConsumer(context, consumerGroup, clientChannelInfo); + if (channel != null) { + ClientChannelInfo clientChannelInfo = new ClientChannelInfo( + channel, + requestHeader.getClientID(), + request.getLanguage(), + request.getVersion()); + this.messagingProcessor.unRegisterConsumer(context, consumerGroup, clientChannelInfo); + } else { + log.warn("unregister consumer failed, channel not exist, may has been removed, consumerGroup={}, channel={}", consumerGroup, ctx.channel()); + } } response.setCode(ResponseCode.SUCCESS); response.setRemark(""); diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java index c6e62487309..10667566aa0 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java @@ -19,6 +19,7 @@ import java.io.File; import java.net.InetAddress; import java.net.UnknownHostException; +import java.time.Duration; public class MessageStoreConfig { @@ -59,6 +60,7 @@ public int getValue() { return value; } + @SuppressWarnings("DuplicatedCode") public static TieredStorageLevel valueOf(int value) { switch (value) { case 1: @@ -91,18 +93,18 @@ public boolean check(TieredStorageLevel targetLevel) { private long tieredStoreConsumeQueueMaxSize = 100 * 1024 * 1024; private int tieredStoreIndexFileMaxHashSlotNum = 5000000; private int tieredStoreIndexFileMaxIndexNum = 5000000 * 4; - // index file will force rolling to next file after idle specified time, default is 3h - private int tieredStoreIndexFileRollingIdleInterval = 3 * 60 * 60 * 1000; + private String tieredMetadataServiceProvider = "org.apache.rocketmq.tieredstore.metadata.DefaultMetadataStore"; private String tieredBackendServiceProvider = "org.apache.rocketmq.tieredstore.provider.MemoryFileSegment"; + // file reserved time, default is 72 hour + private boolean tieredStoreDeleteFileEnable = true; private int tieredStoreFileReservedTime = 72; + private long tieredStoreDeleteFileInterval = Duration.ofHours(1).toMillis(); + // time of forcing commitLog to roll to next file, default is 24 hour private int commitLogRollingInterval = 24; - // rolling will only happen if file segment size is larger than commitcp b LogRollingMinimumSize, default is 128M - private int commitLogRollingMinimumSize = 128 * 1024 * 1024; - // default is 100, unit is millisecond - private int maxCommitJitter = 100; + private int commitLogRollingMinimumSize = 16 * 1024 * 1024; private boolean tieredStoreGroupCommit = true; private int tieredStoreGroupCommitTimeout = 30 * 1000; @@ -112,7 +114,6 @@ public boolean check(TieredStorageLevel targetLevel) { private int tieredStoreGroupCommitSize = 4 * 1024 * 1024; // Cached message count larger than this value will suspend append. default is 10000 private int tieredStoreMaxGroupCommitCount = 10000; - private long tieredStoreMaxFallBehindSize = 128 * 1024 * 1024; private boolean readAheadCacheEnable = true; private int readAheadMessageCountThreshold = 4096; @@ -226,14 +227,6 @@ public void setTieredStoreIndexFileMaxIndexNum(int tieredStoreIndexFileMaxIndexN this.tieredStoreIndexFileMaxIndexNum = tieredStoreIndexFileMaxIndexNum; } - public int getTieredStoreIndexFileRollingIdleInterval() { - return tieredStoreIndexFileRollingIdleInterval; - } - - public void setTieredStoreIndexFileRollingIdleInterval(int tieredStoreIndexFileRollingIdleInterval) { - this.tieredStoreIndexFileRollingIdleInterval = tieredStoreIndexFileRollingIdleInterval; - } - public String getTieredMetadataServiceProvider() { return tieredMetadataServiceProvider; } @@ -250,6 +243,14 @@ public void setTieredBackendServiceProvider(String tieredBackendServiceProvider) this.tieredBackendServiceProvider = tieredBackendServiceProvider; } + public boolean isTieredStoreDeleteFileEnable() { + return tieredStoreDeleteFileEnable; + } + + public void setTieredStoreDeleteFileEnable(boolean tieredStoreDeleteFileEnable) { + this.tieredStoreDeleteFileEnable = tieredStoreDeleteFileEnable; + } + public int getTieredStoreFileReservedTime() { return tieredStoreFileReservedTime; } @@ -258,6 +259,14 @@ public void setTieredStoreFileReservedTime(int tieredStoreFileReservedTime) { this.tieredStoreFileReservedTime = tieredStoreFileReservedTime; } + public long getTieredStoreDeleteFileInterval() { + return tieredStoreDeleteFileInterval; + } + + public void setTieredStoreDeleteFileInterval(long tieredStoreDeleteFileInterval) { + this.tieredStoreDeleteFileInterval = tieredStoreDeleteFileInterval; + } + public int getCommitLogRollingInterval() { return commitLogRollingInterval; } @@ -274,14 +283,6 @@ public void setCommitLogRollingMinimumSize(int commitLogRollingMinimumSize) { this.commitLogRollingMinimumSize = commitLogRollingMinimumSize; } - public int getMaxCommitJitter() { - return maxCommitJitter; - } - - public void setMaxCommitJitter(int maxCommitJitter) { - this.maxCommitJitter = maxCommitJitter; - } - public boolean isTieredStoreGroupCommit() { return tieredStoreGroupCommit; } @@ -322,14 +323,6 @@ public void setTieredStoreMaxGroupCommitCount(int tieredStoreMaxGroupCommitCount this.tieredStoreMaxGroupCommitCount = tieredStoreMaxGroupCommitCount; } - public long getTieredStoreMaxFallBehindSize() { - return tieredStoreMaxFallBehindSize; - } - - public void setTieredStoreMaxFallBehindSize(long tieredStoreMaxFallBehindSize) { - this.tieredStoreMaxFallBehindSize = tieredStoreMaxFallBehindSize; - } - public boolean isReadAheadCacheEnable() { return readAheadCacheEnable; } diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java index 9a25f85a6b8..7b63e16696e 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java @@ -104,6 +104,9 @@ public boolean load() { if (result) { indexService.start(); dispatcher.start(); + storeExecutor.commonExecutor.scheduleWithFixedDelay( + flatFileStore::scheduleDeleteExpireFile, storeConfig.getTieredStoreDeleteFileInterval(), + storeConfig.getTieredStoreDeleteFileInterval(), TimeUnit.MILLISECONDS); } return result; } @@ -457,12 +460,12 @@ public synchronized void shutdown() { if (dispatcher != null) { dispatcher.shutdown(); } - if (flatFileStore != null) { - flatFileStore.shutdown(); - } if (indexService != null) { indexService.shutdown(); } + if (flatFileStore != null) { + flatFileStore.shutdown(); + } if (storeExecutor != null) { storeExecutor.shutdown(); } @@ -473,6 +476,9 @@ public void destroy() { if (next != null) { next.destroy(); } + if (indexService != null) { + indexService.destroy(); + } if (flatFileStore != null) { flatFileStore.destroy(); } diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java index 330872ab9cd..ee06700b8b0 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java @@ -138,9 +138,13 @@ public CompletableFuture doScheduleDispatch(FlatFileInterface flatFile, // If set to max offset here, some written messages may be lost if (!flatFile.isFlatFileInit()) { - currentOffset = Math.max(minOffsetInQueue, - maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize()); + currentOffset = defaultStore.getOffsetInQueueByTime( + topic, queueId, System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(2)); + currentOffset = Math.max(currentOffset, minOffsetInQueue); + currentOffset = Math.min(currentOffset, maxOffsetInQueue); flatFile.initOffset(currentOffset); + log.warn("MessageDispatcher#dispatch init, topic={}, queueId={}, offset={}-{}, current={}", + topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset); return CompletableFuture.completedFuture(true); } diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java index b72ebe86241..7f79dbcd984 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java @@ -39,6 +39,7 @@ import org.apache.rocketmq.tieredstore.file.FlatFileStore; import org.apache.rocketmq.tieredstore.file.FlatMessageFile; import org.apache.rocketmq.tieredstore.index.IndexItem; +import org.apache.rocketmq.tieredstore.index.IndexService; import org.apache.rocketmq.tieredstore.metadata.MetadataStore; import org.apache.rocketmq.tieredstore.metadata.entity.TopicMetadata; import org.apache.rocketmq.tieredstore.util.MessageFormatUtil; @@ -56,15 +57,24 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher { private final MetadataStore metadataStore; private final MessageStoreConfig storeConfig; private final TieredMessageStore messageStore; + private final IndexService indexService; private final FlatFileStore flatFileStore; private final long memoryMaxSize; private final Cache fetcherCache; public MessageStoreFetcherImpl(TieredMessageStore messageStore) { - this.storeConfig = messageStore.getStoreConfig(); + this(messageStore, messageStore.getStoreConfig(), + messageStore.getFlatFileStore(), messageStore.getIndexService()); + } + + public MessageStoreFetcherImpl(TieredMessageStore messageStore, MessageStoreConfig storeConfig, + FlatFileStore flatFileStore, IndexService indexService) { + + this.storeConfig = storeConfig; this.brokerName = storeConfig.getBrokerName(); - this.flatFileStore = messageStore.getFlatFileStore(); + this.flatFileStore = flatFileStore; this.messageStore = messageStore; + this.indexService = indexService; this.metadataStore = flatFileStore.getMetadataStore(); this.memoryMaxSize = (long) (Runtime.getRuntime().maxMemory() * storeConfig.getReadAheadCacheSizeThresholdRate()); @@ -192,7 +202,11 @@ public CompletableFuture getMessageFromCacheAsync( log.debug("MessageFetcher cache miss, group={}, topic={}, queueId={}, offset={}, maxCount={}, lag={}", group, mq.getTopic(), mq.getQueueId(), queueOffset, maxCount, result.getMaxOffset() - result.getNextBeginOffset()); - return fetchMessageThenPutToCache(flatFile, queueOffset, storeConfig.getReadAheadMessageCountThreshold()) + // To optimize the performance of pop consumption + // Pop revive will cause a large number of random reads, + // so the amount of pre-fetch message num needs to be reduced. + int fetchSize = maxCount == 1 ? 32 : storeConfig.getReadAheadMessageCountThreshold(); + return fetchMessageThenPutToCache(flatFile, queueOffset, fetchSize) .thenApply(maxOffset -> getMessageFromCache(flatFile, queueOffset, maxCount, messageFilter)); } @@ -414,8 +428,7 @@ public CompletableFuture queryMessageAsync( return CompletableFuture.completedFuture(new QueryMessageResult()); } - CompletableFuture> future = - messageStore.getIndexService().queryAsync(topic, key, maxCount, begin, end); + CompletableFuture> future = indexService.queryAsync(topic, key, maxCount, begin, end); return future.thenCompose(indexItemList -> { List> futureList = new ArrayList<>(maxCount); diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java index 6ac0939571f..16c05204759 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java @@ -19,6 +19,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; import org.apache.rocketmq.tieredstore.common.FileSegmentType; +import org.apache.rocketmq.tieredstore.provider.FileSegment; import org.apache.rocketmq.tieredstore.provider.FileSegmentFactory; import org.apache.rocketmq.tieredstore.util.MessageFormatUtil; @@ -33,10 +34,19 @@ public FlatCommitLogFile(FileSegmentFactory fileSegmentFactory, String filePath) this.initOffset(0L); } + /** + * Two rules are set here: + * 1. Single file must be saved for more than one day as default. + * 2. Single file must reach the minimum size before switching. + * When calculating storage space, due to the limitation of condition 2, + * the actual usage of storage space may be slightly higher than expected. + */ public boolean tryRollingFile(long interval) { - long timestamp = this.getFileToWrite().getMinTimestamp(); - if (timestamp != Long.MAX_VALUE && - timestamp + interval < System.currentTimeMillis()) { + FileSegment fileSegment = this.getFileToWrite(); + long timestamp = fileSegment.getMinTimestamp(); + if (timestamp != Long.MAX_VALUE && timestamp + interval < System.currentTimeMillis() && + fileSegment.getAppendPosition() >= + fileSegmentFactory.getStoreConfig().getCommitLogRollingMinimumSize()) { this.rollingNewFile(this.getAppendOffset()); return true; } diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatFileStore.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatFileStore.java index f782d099def..70ba2178010 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatFileStore.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatFileStore.java @@ -59,16 +59,6 @@ public boolean load() { try { this.flatFileConcurrentMap.clear(); this.recover(); - this.executor.commonExecutor.scheduleWithFixedDelay(() -> { - for (FlatMessageFile flatFile : deepCopyFlatFileToList()) { - long expiredTimeStamp = System.currentTimeMillis() - - TimeUnit.HOURS.toMillis(flatFile.getFileReservedHours()); - flatFile.destroyExpiredFile(expiredTimeStamp); - if (flatFile.consumeQueue.fileSegmentTable.isEmpty()) { - this.destroyFile(flatFile.getMessageQueue()); - } - } - }, 60, 60, TimeUnit.SECONDS); log.info("FlatFileStore recover finished, total cost={}ms", stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { long costTime = stopwatch.elapsed(TimeUnit.MILLISECONDS); @@ -113,6 +103,27 @@ public CompletableFuture recoverAsync(TopicMetadata topicMetadata) { }, executor.bufferCommitExecutor); } + public void scheduleDeleteExpireFile() { + if (!storeConfig.isTieredStoreDeleteFileEnable()) { + return; + } + Stopwatch stopwatch = Stopwatch.createStarted(); + ImmutableList fileList = this.deepCopyFlatFileToList(); + for (FlatMessageFile flatFile : fileList) { + flatFile.getFileLock().lock(); + try { + flatFile.destroyExpiredFile(System.currentTimeMillis() - + TimeUnit.HOURS.toMillis(flatFile.getFileReservedHours())); + } catch (Exception e) { + log.error("FlatFileStore delete expire file error", e); + } finally { + flatFile.getFileLock().unlock(); + } + } + log.info("FlatFileStore schedule delete expired file, count={}, cost={}ms", + fileList.size(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); + } + public MetadataStore getMetadataStore() { return metadataStore; } diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexService.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexService.java index 70c36c88042..a4ea7e78a85 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexService.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexService.java @@ -52,6 +52,9 @@ AppendResult putKey( */ CompletableFuture> queryAsync(String topic, String key, int maxCount, long beginTime, long endTime); + default void forceUpload() { + } + /** * Shutdown the index service. */ diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java index 9e53d97b98c..020b9f3b068 100644 --- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java +++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java @@ -42,6 +42,8 @@ import org.apache.rocketmq.store.logfile.MappedFile; import org.apache.rocketmq.tieredstore.MessageStoreConfig; import org.apache.rocketmq.tieredstore.common.AppendResult; +import org.apache.rocketmq.tieredstore.exception.TieredStoreErrorCode; +import org.apache.rocketmq.tieredstore.exception.TieredStoreException; import org.apache.rocketmq.tieredstore.file.FlatAppendFile; import org.apache.rocketmq.tieredstore.file.FlatFileFactory; import org.apache.rocketmq.tieredstore.provider.FileSegment; @@ -66,18 +68,29 @@ public class IndexStoreService extends ServiceThread implements IndexService { private final AtomicLong compactTimestamp; private final String filePath; private final FlatFileFactory fileAllocator; + private final boolean autoCreateNewFile; - private IndexFile currentWriteFile; - private FlatAppendFile flatAppendFile; + private volatile IndexFile currentWriteFile; + private volatile FlatAppendFile flatAppendFile; public IndexStoreService(FlatFileFactory flatFileFactory, String filePath) { + this(flatFileFactory, filePath, true); + } + + public IndexStoreService(FlatFileFactory flatFileFactory, String filePath, boolean autoCreateNewFile) { this.storeConfig = flatFileFactory.getStoreConfig(); this.filePath = filePath; this.fileAllocator = flatFileFactory; this.timeStoreTable = new ConcurrentSkipListMap<>(); this.compactTimestamp = new AtomicLong(0L); this.readWriteLock = new ReentrantReadWriteLock(); + this.autoCreateNewFile = autoCreateNewFile; + } + + @Override + public void start() { this.recover(); + super.start(); } private void doConvertOldFormatFile(String filePath) { @@ -131,12 +144,14 @@ private void recover() { } } - if (this.timeStoreTable.isEmpty()) { + if (this.autoCreateNewFile && this.timeStoreTable.isEmpty()) { this.createNewIndexFile(System.currentTimeMillis()); } - this.currentWriteFile = this.timeStoreTable.lastEntry().getValue(); - this.setCompactTimestamp(this.timeStoreTable.firstKey() - 1); + if (!this.timeStoreTable.isEmpty()) { + this.currentWriteFile = this.timeStoreTable.lastEntry().getValue(); + this.setCompactTimestamp(this.timeStoreTable.firstKey() - 1); + } // recover remote this.flatAppendFile = fileAllocator.createFlatFileForIndexFile(filePath); @@ -206,7 +221,7 @@ public AppendResult putKey( log.error("IndexStoreService put key three times return error, topic: {}, topicId: {}, " + "queueId: {}, keySize: {}, timestamp: {}", topic, topicId, queueId, keySet.size(), timestamp); - return AppendResult.UNKNOWN_ERROR; + return AppendResult.SUCCESS; } @Override @@ -252,6 +267,30 @@ public CompletableFuture> queryAsync( return future; } + @Override + public void forceUpload() { + try { + readWriteLock.writeLock().lock(); + if (this.currentWriteFile == null) { + log.warn("IndexStoreService no need force upload current write file"); + return; + } + // note: current file has been shutdown before + IndexStoreFile lastFile = new IndexStoreFile(storeConfig, currentWriteFile.getTimestamp()); + if (this.doCompactThenUploadFile(lastFile)) { + this.setCompactTimestamp(lastFile.getTimestamp()); + } else { + throw new TieredStoreException( + TieredStoreErrorCode.UNKNOWN, "IndexStoreService force compact current file error"); + } + } catch (Exception e) { + log.error("IndexStoreService force upload error", e); + throw new RuntimeException(e); + } finally { + readWriteLock.writeLock().lock(); + } + } + public boolean doCompactThenUploadFile(IndexFile indexFile) { if (IndexFile.IndexStatusEnum.UPLOAD.equals(indexFile.getFileStatus())) { log.error("IndexStoreService file status not correct, so skip, timestamp: {}, status: {}", @@ -359,6 +398,9 @@ public void shutdown() { for (Map.Entry entry : timeStoreTable.entrySet()) { entry.getValue().shutdown(); } + if (!autoCreateNewFile) { + this.forceUpload(); + } this.timeStoreTable.clear(); } catch (Exception e) { log.error("IndexStoreService shutdown error", e); @@ -373,7 +415,6 @@ public void run() { long expireTimestamp = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(storeConfig.getTieredStoreFileReservedTime()); this.destroyExpiredFile(expireTimestamp); - IndexFile indexFile = this.getNextSealedFile(); if (indexFile != null) { if (this.doCompactThenUploadFile(indexFile)) { diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java index 8ac7e068a76..92e989e596f 100644 --- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java +++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java @@ -99,6 +99,7 @@ public void dispatchFromCommitLogTest() throws Exception { messageStore = Mockito.mock(TieredMessageStore.class); IndexService indexService = new IndexStoreService(new FlatFileFactory(metadataStore, storeConfig), storePath); + indexService.start(); Mockito.when(messageStore.getDefaultStore()).thenReturn(defaultStore); Mockito.when(messageStore.getStoreConfig()).thenReturn(storeConfig); Mockito.when(messageStore.getStoreExecutor()).thenReturn(executor); diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java index 1e912690b2f..0fbf5a6a843 100644 --- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java +++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java @@ -77,6 +77,7 @@ public void tryRollingFileTest() throws InterruptedException { byteBuffer.putLong(MessageFormatUtil.QUEUE_OFFSET_POSITION, i); Assert.assertEquals(AppendResult.SUCCESS, flatFile.append(byteBuffer, i)); TimeUnit.MILLISECONDS.sleep(2); + storeConfig.setCommitLogRollingMinimumSize(byteBuffer.remaining()); Assert.assertTrue(flatFile.tryRollingFile(1)); } Assert.assertEquals(4, flatFile.fileSegmentTable.size()); diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java index ec55a028bb9..fb563f7c6c2 100644 --- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java +++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java @@ -93,6 +93,7 @@ public void shutdown() { @Test public void basicServiceTest() throws InterruptedException { indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); for (int i = 0; i < 50; i++) { Assert.assertEquals(AppendResult.SUCCESS, indexService.putKey( TOPIC_NAME, TOPIC_ID, QUEUE_ID, KEY_SET, i * 100, MESSAGE_SIZE, System.currentTimeMillis())); @@ -105,6 +106,7 @@ public void basicServiceTest() throws InterruptedException { @Test public void doConvertOldFormatTest() throws IOException { indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); long timestamp = indexService.getTimeStoreTable().firstKey(); Assert.assertEquals(AppendResult.SUCCESS, indexService.putKey( TOPIC_NAME, TOPIC_ID, QUEUE_ID, KEY_SET, MESSAGE_OFFSET, MESSAGE_SIZE, timestamp)); @@ -116,6 +118,7 @@ public void doConvertOldFormatTest() throws IOException { mappedFile.shutdown(10 * 1000); indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); ConcurrentSkipListMap timeStoreTable = indexService.getTimeStoreTable(); Assert.assertEquals(1, timeStoreTable.size()); Assert.assertEquals(Long.valueOf(timestamp), timeStoreTable.firstKey()); @@ -129,6 +132,7 @@ public void concurrentPutTest() throws InterruptedException { storeConfig.setTieredStoreIndexFileMaxHashSlotNum(500); storeConfig.setTieredStoreIndexFileMaxIndexNum(2000); indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); long timestamp = System.currentTimeMillis(); // first item is invalid @@ -205,6 +209,7 @@ public void runServiceTest() throws InterruptedException { @Test public void restartServiceTest() throws InterruptedException { indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); for (int i = 0; i < 20; i++) { AppendResult result = indexService.putKey( TOPIC_NAME, TOPIC_ID, QUEUE_ID, Collections.singleton(String.valueOf(i)), @@ -214,10 +219,10 @@ public void restartServiceTest() throws InterruptedException { } long timestamp = indexService.getTimeStoreTable().firstKey(); indexService.shutdown(); - indexService = new IndexStoreService(fileAllocator, filePath); - Assert.assertEquals(timestamp, indexService.getTimeStoreTable().firstKey().longValue()); + indexService = new IndexStoreService(fileAllocator, filePath); indexService.start(); + Assert.assertEquals(timestamp, indexService.getTimeStoreTable().firstKey().longValue()); await().atMost(Duration.ofMinutes(1)).pollInterval(Duration.ofSeconds(1)).until(() -> { ArrayList files = new ArrayList<>(indexService.getTimeStoreTable().values()); return IndexFile.IndexStatusEnum.UPLOAD.equals(files.get(0).getFileStatus()); @@ -225,6 +230,7 @@ public void restartServiceTest() throws InterruptedException { indexService.shutdown(); indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); Assert.assertEquals(timestamp, indexService.getTimeStoreTable().firstKey().longValue()); Assert.assertEquals(2, indexService.getTimeStoreTable().size()); Assert.assertEquals(IndexFile.IndexStatusEnum.UPLOAD, @@ -235,6 +241,7 @@ public void restartServiceTest() throws InterruptedException { public void queryFromFileTest() throws InterruptedException, ExecutionException { long timestamp = System.currentTimeMillis(); indexService = new IndexStoreService(fileAllocator, filePath); + indexService.start(); // three files, echo contains 19 items for (int i = 0; i < 3; i++) {