opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 05:44:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3708] Both CQ and BCQ need to be supported in DefaultMessageStore. (#3712)

Browse Source 
* [RIP-26] Both CQ and BCQ will be supported in DefaultMessageStore.

* [RIP-26] Both CQ and BCQ will be supported in DefaultMessageStore.
This commit is contained in:
Hongjian Fei
2022-02-08 10:38:41 +08:00
committed by RongtongJin
parent 5ce1b88cf0
commit e8cf133874
48 changed files with 906 additions and 3141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+2 -9
View File
@@ -96,11 +96,9 @@ import org.apache.rocketmq.srvutil.FileWatchService;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.StreamMessageStore;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.dledger.DLedgerCommitLog;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.store.stats.BrokerStats;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -298,14 +296,9 @@ public class BrokerController {
if (result) {
try {
MessageStore messageStore;
if (Objects.equals(CQType.BatchCQ.toString(), this.messageStoreConfig.getDefaultCQType())) {
messageStore = new StreamMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig);
} else {
messageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig);
}
this.messageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig);
((DefaultMessageStore) this.messageStore).setTopicConfigTable(topicConfigManager.getTopicConfigTable());
this.messageStore = messageStore;
if (messageStoreConfig.isEnableDLegerCommitLog()) {
DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
((DLedgerCommitLog) messageStore.getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
-1
View File
@@ -33,7 +33,6 @@ import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.admin.ConsumeStats;
broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java
-2
View File
@@ -32,12 +32,10 @@ import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetResponseHe
import org.apache.rocketmq.common.rpc.RpcClientUtils;
import org.apache.rocketmq.common.rpc.RpcRequest;
import org.apache.rocketmq.common.rpc.RpcResponse;
import org.apache.rocketmq.common.rpc.TopicQueueRequestHeader;
import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
import org.apache.rocketmq.common.sysflag.PullSysFlag;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
+3 -4
View File
@@ -18,9 +18,9 @@ package org.apache.rocketmq.broker.processor;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadLocalRandom;
@@ -42,7 +42,6 @@ import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
@@ -56,6 +55,7 @@ import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.netty.RemotingResponseCallback;
@@ -63,7 +63,6 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.MessageExtBatch;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.StoreUtil;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -684,7 +683,7 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
CompletableFuture<PutMessageResult> putMessageResult;
if (StoreUtil.isStreamMode(this.brokerController.getMessageStore()) && MessageClientIDSetter.getUniqID(messageExtBatch) != null) {
if (QueueTypeUtils.isBatchCq(Optional.of(topicConfig)) && MessageClientIDSetter.getUniqID(messageExtBatch) != null) {
// newly introduced inner-batch message
messageExtBatch.setSysFlag(messageExtBatch.getSysFlag() | MessageSysFlag.NEED_UNWRAP_FLAG);
messageExtBatch.setSysFlag(messageExtBatch.getSysFlag() | MessageSysFlag.INNER_BATCH_FLAG);
broker/src/test/java/org/apache/rocketmq/broker/topic/TopicConfigManagerTest.java
+5 -4
View File
@@ -24,9 +24,9 @@ import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.BooleanAttribute;
import org.apache.rocketmq.common.attribute.EnumAttribute;
import org.apache.rocketmq.common.attribute.LongRangeAttribute;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.store.util.QueueTypeUtils;
import org.apache.rocketmq.common.attribute.CQType;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -37,6 +37,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import static com.google.common.collect.Sets.newHashSet;
import static java.util.Arrays.asList;
@@ -123,7 +124,7 @@ public class TopicConfigManagerTest {
@Test
public void testAddWrongValueOnCreating() {
Map<String, String> attributes = new HashMap<>();
attributes.put("+" + TopicAttributes.QUEUE_TYPE.getName(), "wrong-value");
attributes.put("+" + TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName(), "wrong-value");
TopicConfig topicConfig = new TopicConfig();
topicConfig.setTopicName("new-topic");
@@ -300,7 +301,7 @@ public class TopicConfigManagerTest {
topicConfigManager.updateTopicConfig(topicConfig);
TopicConfig topicConfigUpdated = topicConfigManager.getTopicConfigTable().get(topic);
Assert.assertEquals(CQType.SimpleCQ, QueueTypeUtils.getCQType(topicConfigUpdated));
Assert.assertEquals(CQType.SimpleCQ, QueueTypeUtils.getCQType(Optional.of(topicConfigUpdated)));
Assert.assertEquals("true", topicConfigUpdated.getAttributes().get(unchangeable));
}
client/src/main/java/org/apache/rocketmq/client/MQAdmin.java
+8 -6
View File
@@ -22,29 +22,31 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Map;
/**
* Base interface for MQ management
*/
public interface MQAdmin {
/**
* Creates an topic
*
* @param key accesskey
* @param key accesskey
* @param newTopic topic name
* @param queueNum topic's queue number
* @param attributes
*/
void createTopic(final String key, final String newTopic, final int queueNum)
void createTopic(final String key, final String newTopic, final int queueNum, Map<String, String> attributes)
throws MQClientException;
/**
* Creates an topic
*
* @param key accesskey
* @param key accesskey
* @param newTopic topic name
* @param queueNum topic's queue number
* @param topicSysFlag topic system flag
* @param attributes
*/
void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes)
throws MQClientException;
/**
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
+4 -3
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.client.consumer;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.QueryResult;
@@ -124,8 +125,8 @@ public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsume
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0);
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0, null);
}
/**
@@ -133,7 +134,7 @@ public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsume
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
this.defaultMQPullConsumerImpl.createTopic(key, withNamespace(newTopic), queueNum, topicSysFlag);
}
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
+3 -3
View File
@@ -431,8 +431,8 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0);
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0, null);
}
@Override
@@ -448,7 +448,7 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
this.defaultMQPushConsumerImpl.createTopic(key, withNamespace(newTopic), queueNum, topicSysFlag);
}
client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
+4 -2
View File
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -76,10 +77,10 @@ public class MQAdminImpl {
}
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, newTopic, queueNum, 0);
createTopic(key, newTopic, queueNum, 0, null);
}
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
try {
Validators.checkTopic(newTopic);
Validators.isSystemTopic(newTopic);
@@ -100,6 +101,7 @@ public class MQAdminImpl {
topicConfig.setReadQueueNums(queueNum);
topicConfig.setWriteQueueNums(queueNum);
topicConfig.setTopicSysFlag(topicSysFlag);
topicConfig.setAttributes(attributes);
boolean createOK = false;
for (int i = 0; i < 5; i++) {
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+1 -1
View File
@@ -100,7 +100,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
this.isRunning();
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag, null);
}
private void isRunning() throws MQClientException {
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+1 -1
View File
@@ -184,7 +184,7 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
}
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag, null);
}
public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClientException {
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+1 -1
View File
@@ -447,7 +447,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
Validators.checkTopic(newTopic);
Validators.isSystemTopic(newTopic);
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag, null);
}
private void makeSureStateOK() throws MQClientException {
client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
+6 -3
View File
@@ -19,6 +19,7 @@ package org.apache.rocketmq.client.producer;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
@@ -757,12 +758,13 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @param key accesskey
* @param newTopic topic name
* @param queueNum topic's queue number
* @param attributes
* @throws MQClientException if there is any client error.
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0);
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
createTopic(key, withNamespace(newTopic), queueNum, 0, null);
}
/**
@@ -773,11 +775,12 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @param newTopic topic name
* @param queueNum topic's queue number
* @param topicSysFlag topic system flag
* @param attributes
* @throws MQClientException if there is any client error.
*/
@Deprecated
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
this.defaultMQProducerImpl.createTopic(key, withNamespace(newTopic), queueNum, topicSysFlag);
}
common/src/main/java/org/apache/rocketmq/common/TopicAttributes.java
+2 -2
View File
@@ -25,7 +25,7 @@ import java.util.Map;
import static com.google.common.collect.Sets.newHashSet;
public class TopicAttributes {
public static final EnumAttribute QUEUE_TYPE = new EnumAttribute(
public static final EnumAttribute QUEUE_TYPE_ATTRIBUTE = new EnumAttribute(
"queue.type",
false,
newHashSet("BatchCQ", "SimpleCQ"),
@@ -35,6 +35,6 @@ public class TopicAttributes {
static {
ALL = new HashMap<>();
ALL.put(QUEUE_TYPE.getName(), QUEUE_TYPE);
ALL.put(QUEUE_TYPE_ATTRIBUTE.getName(), QUEUE_TYPE_ATTRIBUTE);
}
}
store/src/main/java/org/apache/rocketmq/store/queue/CQType.java → common/src/main/java/org/apache/rocketmq/common/attribute/CQType.java
+1 -1
View File
@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.store.queue;
package org.apache.rocketmq.common.attribute;
public enum CQType {
SimpleCQ,
common/src/main/java/org/apache/rocketmq/common/utils/QueueTypeUtils.java
+51
View File
@@ -0,0 +1,51 @@
/*
* 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.utils;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.CQType;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
public class QueueTypeUtils {
public static boolean isBatchCq(Optional<TopicConfig> topicConfig) {
return Objects.equals(CQType.BatchCQ, getCQType(topicConfig));
}
public static CQType getCQType(Optional<TopicConfig> topicConfig) {
if (!topicConfig.isPresent()) {
return CQType.valueOf(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getDefaultValue());
}
String attributeName = TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName();
Map<String, String> attributes = topicConfig.get().getAttributes();
if (attributes == null || attributes.size() == 0) {
return CQType.valueOf(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getDefaultValue());
}
if (attributes.containsKey(attributeName)) {
return CQType.valueOf(attributes.get(attributeName));
} else {
return CQType.valueOf(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getDefaultValue());
}
}
}
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+21 -24
View File
@@ -24,10 +24,10 @@ import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
@@ -44,6 +45,7 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
@@ -51,9 +53,8 @@ import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.HAService;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.store.schedule.ScheduleMessageService;
import org.apache.rocketmq.store.util.QueueTypeUtils;
/**
* Store all metadata downtime for recovery, data protection reliability
@@ -433,8 +434,8 @@ public class CommitLog implements Swappable {
private void setBatchSizeIfNeeded(Map<String, String> propertiesMap, DispatchRequest dispatchRequest) {
if (null != propertiesMap && propertiesMap.containsKey(MessageConst.PROPERTY_INNER_NUM) && propertiesMap.containsKey(MessageConst.PROPERTY_INNER_BASE)) {
dispatchRequest.setMsgBaseOffset(Long.valueOf(propertiesMap.get(MessageConst.PROPERTY_INNER_BASE)));
dispatchRequest.setBatchSize(Short.valueOf(propertiesMap.get(MessageConst.PROPERTY_INNER_NUM)));
dispatchRequest.setMsgBaseOffset(Long.parseLong(propertiesMap.get(MessageConst.PROPERTY_INNER_BASE)));
dispatchRequest.setBatchSize(Short.parseShort(propertiesMap.get(MessageConst.PROPERTY_INNER_NUM)));
}
}
@@ -672,7 +673,7 @@ public class CommitLog implements Swappable {
topicQueueLock.lock(topicQueueKey);
try {
defaultMessageStore.assignOffset(topicQueueKey, msg, getBatchNum(msg));
defaultMessageStore.assignOffset(topicQueueKey, msg, getMessageNum(msg));
PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);
if (encodeResult != null) {
@@ -1031,15 +1032,6 @@ public class CommitLog implements Swappable {
return this.mappedFileQueue.retryDeleteFirstFile(intervalForcibly);
}
public void removeQueueFromTopicQueueTable(final String topic, final int queueId) {
String key = topic + "-" + queueId;
synchronized (this) {
this.defaultMessageStore.removeOffsetTable(key);
}
log.info("removeQueueFromTopicQueueTable OK Topic: {} QueueId: {}", topic, queueId);
}
public void checkSelf() {
mappedFileQueue.checkSelf();
}
@@ -1058,19 +1050,24 @@ public class CommitLog implements Swappable {
return diff;
}
protected short getBatchNum(MessageExtBrokerInner msgInner) {
short batchNum = 1;
protected short getMessageNum(MessageExtBrokerInner msgInner) {
short messageNum = 1;
// IF inner batch, build batchQueueOffset and batchNum property.
CQType cqType = QueueTypeUtils.getCQType(defaultMessageStore);
if (MessageSysFlag.check(msgInner.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG) || CQType.BatchCQ.equals(cqType)) {
CQType cqType = getCqType(msgInner);
if (MessageSysFlag.check(msgInner.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG) || CQType.BatchCQ.equals(cqType)) {
if (msgInner.getProperty(MessageConst.PROPERTY_INNER_NUM) != null) {
batchNum = Short.parseShort(msgInner.getProperty(MessageConst.PROPERTY_INNER_NUM));
batchNum = batchNum >= 1 ? batchNum : 1;
messageNum = Short.parseShort(msgInner.getProperty(MessageConst.PROPERTY_INNER_NUM));
messageNum = messageNum >= 1 ? messageNum : 1;
}
}
return batchNum;
return messageNum;
}
private CQType getCqType(MessageExtBrokerInner msgInner) {
Optional<TopicConfig> topicConfig = this.defaultMessageStore.getTopicConfig(msgInner.getTopic());
return QueueTypeUtils.getCQType(topicConfig);
}
abstract class FlushCommitLogService extends ServiceThread {
@@ -1488,7 +1485,7 @@ public class CommitLog implements Swappable {
Long queueOffset = msgInner.getQueueOffset();
// this msg maybe a inner-batch msg.
short batchNum = getBatchNum(msgInner);
short messageNum = getMessageNum(msgInner);
// Transaction messages that require special handling
final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
@@ -1545,7 +1542,7 @@ public class CommitLog implements Swappable {
CommitLog.this.getMessageStore().getPerfCounter().endTick("WRITE_MEMORY_TIME_MS");
msgInner.setEncodedBuff(null);
return new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgIdSupplier,
msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills, batchNum);
msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills, messageNum);
}
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+16 -3
View File
@@ -18,6 +18,7 @@ package org.apache.rocketmq.store;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
@@ -25,10 +26,11 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.FileQueueLifeCycle;
import org.apache.rocketmq.store.queue.QueueOffsetAssigner;
import org.apache.rocketmq.store.queue.ReferredIterator;
public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
@@ -37,7 +39,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
public static final int CQ_STORE_UNIT_SIZE = 20;
private static final InternalLogger LOG_ERROR = InternalLoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
private final DefaultMessageStore defaultMessageStore;
private final MessageStore defaultMessageStore;
private final MappedFileQueue mappedFileQueue;
private final String topic;
@@ -55,7 +57,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
final int queueId,
final String storePath,
final int mappedFileSize,
final DefaultMessageStore defaultMessageStore) {
final MessageStore defaultMessageStore) {
this.storePath = storePath;
this.mappedFileSize = mappedFileSize;
this.defaultMessageStore = defaultMessageStore;
@@ -438,6 +440,17 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
}
@Override
public void assignQueueOffset(QueueOffsetAssigner queueOffsetAssigner, MessageExtBrokerInner msg, short messageNum) {
String topicQueueKey = getTopic() + "-" + getQueueId();
HashMap<String, Long> topicQueueTable = queueOffsetAssigner.getTopicQueueTable();
long topicOffset = topicQueueTable.computeIfAbsent(topicQueueKey, k -> 0L);
topicQueueTable.put(topicQueueKey, topicOffset + messageNum);
msg.setQueueOffset(topicOffset);
}
private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
final long cqOffset) {
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+51 -172
View File
@@ -32,9 +32,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -47,13 +47,16 @@ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.SystemClock;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
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.running.RunningStats;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
@@ -74,8 +77,6 @@ import org.apache.rocketmq.store.schedule.ScheduleMessageService;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.store.util.PerfCounter;
import static java.lang.String.format;
public class DefaultMessageStore implements MessageStore {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
@@ -87,10 +88,6 @@ public class DefaultMessageStore implements MessageStore {
private final ConsumeQueueStore consumeQueueStore;
private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface>> consumeQueueTable;
protected HashMap<String/* topic-queueid */, Long/* offset */> topicQueueTable = new HashMap<>(1024);
private final FlushConsumeQueueService flushConsumeQueueService;
private final CleanCommitLogService cleanCommitLogService;
@@ -154,8 +151,7 @@ public class DefaultMessageStore implements MessageStore {
} else {
this.commitLog = new CommitLog(this);
}
this.consumeQueueTable = new ConcurrentHashMap<>(32);
this.consumeQueueStore = new ConsumeQueueStore(this, this.messageStoreConfig, this.consumeQueueTable);
this.consumeQueueStore = new ConsumeQueueStore(this, this.messageStoreConfig);
this.flushConsumeQueueService = new FlushConsumeQueueService();
this.cleanCommitLogService = new CleanCommitLogService();
@@ -194,13 +190,7 @@ public class DefaultMessageStore implements MessageStore {
@Override
public void truncateDirtyLogicFiles(long phyOffset) {
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.consumeQueueTable;
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.consumeQueueStore.truncateDirtyLogicFiles(logic, phyOffset);
}
}
this.consumeQueueStore.truncateDirty(phyOffset);
}
/**
@@ -211,7 +201,6 @@ public class DefaultMessageStore implements MessageStore {
boolean result = true;
try {
long start = System.currentTimeMillis();
boolean lastExitOK = !this.isTempFileExist();
log.info("last shutdown {}, root dir: {}", lastExitOK ? "normally" : "abnormally", messageStoreConfig.getStorePathRootDir());
@@ -219,7 +208,7 @@ public class DefaultMessageStore implements MessageStore {
result = result && this.commitLog.load();
// load Consume Queue
result = result && this.loadConsumeQueue();
result = result && this.consumeQueueStore.load();
if (result) {
this.storeCheckpoint =
@@ -269,7 +258,7 @@ public class DefaultMessageStore implements MessageStore {
* 4. Make sure the fall-behind messages to be dispatched before starting the commitlog, especially when the broker role are automatically changed.
*/
long maxPhysicalPosInLogicQueue = commitLog.getMinOffset();
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.getConsumeQueueTable().values()) {
for (ConsumeQueueInterface logic : maps.values()) {
if (logic.getMaxPhysicOffset() > maxPhysicalPosInLogicQueue) {
maxPhysicalPosInLogicQueue = logic.getMaxPhysicOffset();
@@ -387,11 +376,7 @@ public class DefaultMessageStore implements MessageStore {
@Override
public void destroyLogics() {
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.consumeQueueStore.destroy(logic);
}
}
this.consumeQueueStore.destroy();
}
private PutMessageStatus checkMessage(MessageExtBrokerInner msg) {
@@ -464,6 +449,20 @@ public class DefaultMessageStore implements MessageStore {
return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));
}
if (msg.getProperties().containsKey(MessageConst.PROPERTY_INNER_NUM)
&& !MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
log.warn("[BUG]The message had property {} but is not an inner batch", MessageConst.PROPERTY_INNER_NUM);
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null));
}
if (MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
Optional<TopicConfig> topicConfig = this.getTopicConfig(msg.getTopic());
if (!QueueTypeUtils.isBatchCq(topicConfig)) {
log.error("[BUG]The message is an inner batch but cq type is not batch cq");
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null));
}
}
long beginTime = this.getSystemClock().now();
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
@@ -649,8 +648,7 @@ public class DefaultMessageStore implements MessageStore {
break;
}
if (this.isTheBatchFull(sizePy, maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(),
isInDisk)) {
if (this.isTheBatchFull(sizePy, cqUnit.getBatchNum(), maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInDisk)) {
break;
}
@@ -756,7 +754,7 @@ public class DefaultMessageStore implements MessageStore {
return logic.getMaxOffsetInQueue();
}
} else {
Long offset = this.topicQueueTable.get(topic + "-" + queueId);
Long offset = this.consumeQueueStore.getMaxOffset(topic, queueId);
if (offset != null) {
return offset;
}
@@ -800,7 +798,11 @@ public class DefaultMessageStore implements MessageStore {
public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) {
ConsumeQueueInterface logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
return logic.getOffsetInQueueByTime(timestamp);
long resultOffset = logic.getOffsetInQueueByTime(timestamp);
// Make sure the result offset is in valid range.
resultOffset = Math.max(resultOffset, logic.getMinOffsetInQueue());
resultOffset = Math.min(resultOffset, logic.getMaxOffsetInQueue());
return resultOffset;
}
return 0;
@@ -1071,7 +1073,7 @@ public class DefaultMessageStore implements MessageStore {
@Override
public int cleanUnusedTopic(Set<String> topics) {
Iterator<Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
Iterator<Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.getConsumeQueueTable().entrySet().iterator();
while (it.hasNext()) {
Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
String topic = next.getKey();
@@ -1086,7 +1088,7 @@ public class DefaultMessageStore implements MessageStore {
cq.getQueueId()
);
this.commitLog.removeQueueFromTopicQueueTable(cq.getTopic(), cq.getQueueId());
this.consumeQueueStore.removeTopicQueueTable(cq.getTopic(), cq.getQueueId());
}
it.remove();
@@ -1105,45 +1107,7 @@ public class DefaultMessageStore implements MessageStore {
public void cleanExpiredConsumerQueue() {
long minCommitLogOffset = this.commitLog.getMinOffset();
Iterator<Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
String topic = next.getKey();
if (!topic.equals(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC)) {
ConcurrentMap<Integer, ConsumeQueueInterface> queueTable = next.getValue();
Iterator<Entry<Integer, ConsumeQueueInterface>> itQT = queueTable.entrySet().iterator();
while (itQT.hasNext()) {
Entry<Integer, ConsumeQueueInterface> nextQT = itQT.next();
long maxCLOffsetInConsumeQueue = nextQT.getValue().getLastOffset();
if (maxCLOffsetInConsumeQueue == -1) {
log.warn("maybe ConsumeQueue was created just now. topic={} queueId={} maxPhysicOffset={} minLogicOffset={}.",
nextQT.getValue().getTopic(),
nextQT.getValue().getQueueId(),
nextQT.getValue().getMaxPhysicOffset(),
nextQT.getValue().getMinLogicOffset());
} else if (maxCLOffsetInConsumeQueue < minCommitLogOffset) {
log.info(
"cleanExpiredConsumerQueue: {} {} consumer queue destroyed, minCommitLogOffset: {} maxCLOffsetInConsumeQueue: {}",
topic,
nextQT.getKey(),
minCommitLogOffset,
maxCLOffsetInConsumeQueue);
DefaultMessageStore.this.commitLog.removeQueueFromTopicQueueTable(nextQT.getValue().getTopic(),
nextQT.getValue().getQueueId());
this.consumeQueueStore.destroy(nextQT.getValue());
itQT.remove();
}
}
if (queueTable.isEmpty()) {
log.info("cleanExpiredConsumerQueue: {},topic destroyed", topic);
it.remove();
}
}
}
this.consumeQueueStore.cleanExpired(minCommitLogOffset);
}
public Map<String, Long> getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset,
@@ -1269,13 +1233,13 @@ public class DefaultMessageStore implements MessageStore {
return (maxOffsetPy - offsetPy) > memory;
}
private boolean isTheBatchFull(int sizePy, int maxMsgNums, long maxMsgSize, int bufferTotal, int messageTotal, boolean isInDisk) {
private boolean isTheBatchFull(int sizePy, int unitBatchNum, int maxMsgNums, long maxMsgSize, int bufferTotal, int messageTotal, boolean isInDisk) {
if (0 == bufferTotal || 0 == messageTotal) {
return false;
}
if (maxMsgNums <= messageTotal) {
if (messageTotal + unitBatchNum > maxMsgNums) {
return true;
}
@@ -1392,16 +1356,7 @@ public class DefaultMessageStore implements MessageStore {
private void checkSelf() {
this.commitLog.checkSelf();
Iterator<Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
Iterator<Entry<Integer, ConsumeQueueInterface>> itNext = next.getValue().entrySet().iterator();
while (itNext.hasNext()) {
Entry<Integer, ConsumeQueueInterface> cq = itNext.next();
this.consumeQueueStore.checkSelf(cq.getValue());
}
}
this.consumeQueueStore.checkSelf();
}
private boolean isTempFileExist() {
@@ -1410,53 +1365,6 @@ public class DefaultMessageStore implements MessageStore {
return file.exists();
}
private boolean loadConsumeQueue() {
checkOtherConsumeQueue();
File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
File[] fileTopicList = dirLogic.listFiles();
if (fileTopicList != null) {
for (File fileTopic : fileTopicList) {
String topic = fileTopic.getName();
File[] fileQueueIdList = fileTopic.listFiles();
if (fileQueueIdList != null) {
for (File fileQueueId : fileQueueIdList) {
int queueId;
try {
queueId = Integer.parseInt(fileQueueId.getName());
} catch (NumberFormatException e) {
continue;
}
ConsumeQueueInterface logic = new ConsumeQueue(
topic,
queueId,
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),
this);
this.putConsumeQueue(topic, queueId, logic);
if (!this.consumeQueueStore.load(logic)) {
return false;
}
}
}
}
}
log.info("load logics queue all over, OK");
return true;
}
private void checkOtherConsumeQueue() {
File dirLogic = new File(StorePathConfigHelper.getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
if (dirLogic.exists()) {
throw new RuntimeException(format("Batch consume queue directory: [%s] exist. Can not load consume queue while batch consume queue exists.",
StorePathConfigHelper.getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir())));
}
}
private void recover(final boolean lastExitOK) {
long recoverCqStart = System.currentTimeMillis();
long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();
@@ -1485,43 +1393,13 @@ public class DefaultMessageStore implements MessageStore {
return transientStorePool;
}
private void putConsumeQueue(final String topic, final int queueId, final ConsumeQueueInterface consumeQueue) {
ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface> map = this.consumeQueueTable.get(topic);
if (null == map) {
map = new ConcurrentHashMap<Integer/* queueId */, ConsumeQueueInterface>();
map.put(queueId, consumeQueue);
this.consumeQueueTable.put(topic, map);
} else {
map.put(queueId, consumeQueue);
}
}
private long recoverConsumeQueue() {
long maxPhysicOffset = -1;
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.consumeQueueStore.recover(logic);
if (logic.getMaxPhysicOffset() > maxPhysicOffset) {
maxPhysicOffset = logic.getMaxPhysicOffset();
}
}
}
return maxPhysicOffset;
return this.consumeQueueStore.recover();
}
public void recoverTopicQueueTable() {
HashMap<String/* topic-queueid */, Long/* offset */> table = new HashMap<String, Long>(1024);
long minPhyOffset = this.commitLog.getMinOffset();
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
String key = logic.getTopic() + "-" + logic.getQueueId();
table.put(key, logic.getMaxOffsetInQueue());
this.consumeQueueStore.correctMinOffset(logic, minPhyOffset);
}
}
this.topicQueueTable = table;
this.consumeQueueStore.recoverOffsetTable(minPhyOffset);
}
@Override
@@ -1539,7 +1417,7 @@ public class DefaultMessageStore implements MessageStore {
}
public ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> getConsumeQueueTable() {
return consumeQueueTable;
return consumeQueueStore.getConsumeQueueTable();
}
@Override
@@ -1569,8 +1447,7 @@ public class DefaultMessageStore implements MessageStore {
}
public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
ConsumeQueueInterface cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
this.consumeQueueStore.putMessagePositionInfoWrapper(cq, dispatchRequest);
this.consumeQueueStore.putMessagePositionInfoWrapper(dispatchRequest);
}
@Override
@@ -1606,7 +1483,7 @@ public class DefaultMessageStore implements MessageStore {
@Override
public ConsumeQueueInterface getConsumeQueue(String topic, int queueId) {
ConcurrentMap<Integer, ConsumeQueueInterface> map = consumeQueueTable.get(topic);
ConcurrentMap<Integer, ConsumeQueueInterface> map = this.getConsumeQueueTable().get(topic);
if (map == null) {
return null;
}
@@ -1656,19 +1533,21 @@ public class DefaultMessageStore implements MessageStore {
}
@Override
public void assignOffset(String topicQueueKey, MessageExtBrokerInner msg, short batchNum) {
public void assignOffset(String topicQueueKey, MessageExtBrokerInner msg, short messageNum) {
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
long topicOffset = this.topicQueueTable.computeIfAbsent(topicQueueKey, k -> 0L);
msg.setQueueOffset(topicOffset);
this.topicQueueTable.put(topicQueueKey, topicOffset + batchNum);
this.consumeQueueStore.assignQueueOffset(msg, messageNum);
}
}
@Override
public void removeOffsetTable(String topicQueueKey) {
this.topicQueueTable.remove(topicQueueKey);
public Optional<TopicConfig> getTopicConfig(String topic) {
return this.consumeQueueStore.getTopicConfig(topic);
}
public void setTopicConfigTable(ConcurrentMap<String, TopicConfig> topicConfigTable) {
this.consumeQueueStore.setTopicConfigTable(topicConfigTable);
}
class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {
@@ -1975,7 +1854,7 @@ public class DefaultMessageStore implements MessageStore {
if (minOffset > this.lastPhysicalMinOffset) {
this.lastPhysicalMinOffset = minOffset;
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.consumeQueueTable;
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.getConsumeQueueTable();
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
@@ -2021,7 +1900,7 @@ public class DefaultMessageStore implements MessageStore {
logicsMsgTimestamp = DefaultMessageStore.this.getStoreCheckpoint().getLogicsMsgTimestamp();
}
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.consumeQueueTable;
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.getConsumeQueueTable();
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
for (ConsumeQueueInterface cq : maps.values()) {
store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
+8 -8
View File
@@ -171,8 +171,8 @@ public class MappedFileQueue implements Swappable {
return true;
}
try {
MappedFile mappedFile = new DefaultMappedFile(file.getPath(), mappedFileSize);
try {
MappedFile mappedFile = new DefaultMappedFile(file.getPath(), mappedFileSize);
mappedFile.setWrotePosition(this.mappedFileSize);
mappedFile.setFlushedPosition(this.mappedFileSize);
@@ -234,13 +234,13 @@ public class MappedFileQueue implements Swappable {
if (this.allocateMappedFileService != null) {
mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
nextNextFilePath, this.mappedFileSize);
} else {
try {
mappedFile = new DefaultMappedFile(nextFilePath, this.mappedFileSize);
} catch (IOException e) {
log.error("create mappedFile exception", e);
}
} else {
try {
mappedFile = new DefaultMappedFile(nextFilePath, this.mappedFileSize);
} catch (IOException e) {
log.error("create mappedFile exception", e);
}
}
if (mappedFile != null) {
if (this.mappedFiles.isEmpty()) {
store/src/main/java/org/apache/rocketmq/store/MessageStore.java
+11 -6
View File
@@ -18,10 +18,12 @@ package org.apache.rocketmq.store;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.common.SystemClock;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
@@ -562,16 +564,19 @@ public interface MessageStore {
boolean isSyncMaster();
/**
* assign an queue offset and increase it.
* Assign an queue offset and increase it.
* If there is a race condition, you need to lock/unlock this method yourself.
*
* @param topicQueueKey topic-queue key
* @param msg message
* @param batchNum batch num
* @param messageNum message num
*/
void assignOffset(String topicQueueKey, MessageExtBrokerInner msg, short batchNum);
void assignOffset(String topicQueueKey, MessageExtBrokerInner msg, short messageNum);
/**
* remove offset table
* @param topicQueueKey topic-queue key
* get topic config
* @param topic topic name
* @return topic config info
*/
void removeOffsetTable(String topicQueueKey);
Optional<TopicConfig> getTopicConfig(String topic);
}
store/src/main/java/org/apache/rocketmq/store/PutMessageContext.java
+28 -27
View File
@@ -8,40 +8,41 @@
*
* 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.
* 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.store;
public class PutMessageContext {
private String topicQueueTableKey;
private long[] phyPos;
private int batchSize;
private String topicQueueTableKey;
private long[] phyPos;
private int batchSize;
public PutMessageContext(String topicQueueTableKey) {
this.topicQueueTableKey = topicQueueTableKey;
}
public PutMessageContext(String topicQueueTableKey) {
this.topicQueueTableKey = topicQueueTableKey;
}
public String getTopicQueueTableKey() {
return topicQueueTableKey;
}
public String getTopicQueueTableKey() {
return topicQueueTableKey;
}
public long[] getPhyPos() {
return phyPos;
}
public long[] getPhyPos() {
return phyPos;
}
public void setPhyPos(long[] phyPos) {
this.phyPos = phyPos;
}
public void setPhyPos(long[] phyPos) {
this.phyPos = phyPos;
}
public int getBatchSize() {
return batchSize;
}
public int getBatchSize() {
return batchSize;
}
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
}
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
}
store/src/main/java/org/apache/rocketmq/store/StoreUtil.java
-4
View File
@@ -44,10 +44,6 @@ public class StoreUtil {
return physicalTotal;
}
public static boolean isStreamMode(MessageStore messageStore) {
return messageStore instanceof StreamMessageStore;
}
public static void fileAppend(MappedFile file, ByteBuffer data) {
boolean success = file.appendMessage(data);
if (!success) {
store/src/main/java/org/apache/rocketmq/store/StreamMessageStore.java
-2573
View File
File diff suppressed because it is too large Load Diff
store/src/main/java/org/apache/rocketmq/store/TopicQueueLock.java
+6 -5
View File
@@ -8,12 +8,13 @@
*
* 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.
* 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.store;
import java.util.ArrayList;
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
-11
View File
@@ -19,7 +19,6 @@ package org.apache.rocketmq.store.config;
import org.apache.rocketmq.common.annotation.ImportantField;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.queue.BatchConsumeQueue;
import org.apache.rocketmq.store.queue.CQType;
import java.io.File;
@@ -206,9 +205,6 @@ public class MessageStoreConfig {
@ImportantField
private boolean enableCleanExpiredOffset = false;
@ImportantField
private String defaultCQType = CQType.SimpleCQ.toString();
private int maxAsyncPutMessageRequests = 5000;
private int pullBatchMaxMessageCount = 160;
@@ -799,13 +795,6 @@ public class MessageStoreConfig {
this.enableCleanExpiredOffset = enableCleanExpiredOffset;
}
public String getDefaultCQType() {
return defaultCQType;
}
public void setDefaultCQType(String defaultCQType) {
this.defaultCQType = defaultCQType;
}
public String getReadOnlyCommitLogStorePaths() {
return readOnlyCommitLogStorePaths;
}
store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
+1 -2
View File
@@ -33,7 +33,6 @@ import io.openmessaging.storage.dledger.utils.DLedgerUtils;
import java.net.Inet6Address;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -431,7 +430,7 @@ public class DLedgerCommitLog extends CommitLog {
String topicQueueKey = msg.getTopic() + "-" + msg.getQueueId();
topicQueueLock.lock(topicQueueKey);
try {
defaultMessageStore.assignOffset(topicQueueKey, msg, getBatchNum(msg));
defaultMessageStore.assignOffset(topicQueueKey, msg, getMessageNum(msg));
encodeResult = this.messageSerializer.serialize(msg);
if (encodeResult.status != AppendMessageStatus.PUT_OK) {
store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java
-1
View File
@@ -18,7 +18,6 @@ package org.apache.rocketmq.store.logfile;
import org.apache.rocketmq.store.AppendMessageCallback;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.MessageExtBatch;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.PutMessageContext;
store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java
+25 -3
View File
@@ -17,11 +17,17 @@
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.MappedFileQueue;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
@@ -29,6 +35,7 @@ import org.apache.rocketmq.store.logfile.MappedFile;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
@@ -386,9 +393,9 @@ public class BatchConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCy
}
@Override
public int deleteExpiredFile(long offset) {
int cnt = this.mappedFileQueue.deleteExpiredFileByOffset(offset, CQ_STORE_UNIT_SIZE);
this.correctMinOffset(offset);
public int deleteExpiredFile(long minCommitLogPos) {
int cnt = this.mappedFileQueue.deleteExpiredFileByOffset(minCommitLogPos, CQ_STORE_UNIT_SIZE);
this.correctMinOffset(minCommitLogPos);
return cnt;
}
@@ -473,6 +480,21 @@ public class BatchConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCy
this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
}
@Override
public void assignQueueOffset(QueueOffsetAssigner queueOffsetAssigner, MessageExtBrokerInner msg, short messageNum) {
HashMap<String, Long> batchTopicQueueTable = queueOffsetAssigner.getBatchTopicQueueTable();
String topicQueueKey = getTopic() + "-" + getQueueId();
Long topicOffset = batchTopicQueueTable.computeIfAbsent(topicQueueKey, k -> 0L);
if (MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_BASE, String.valueOf(topicOffset));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
}
msg.setQueueOffset(topicOffset);
batchTopicQueueTable.put(topicQueueKey, topicOffset + messageNum);
}
boolean putBatchMessagePositionInfo(final long offset, final int size, final long tagsCode, final long storeTime,
final long msgBaseOffset, final short batchSize) {
store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java
+2
View File
@@ -17,6 +17,8 @@
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.attribute.CQType;
public interface ConsumeQueueInterface {
/**
* Get the topic name
store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStore.java
+280 -10
View File
@@ -16,27 +16,49 @@
*/
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.StoreUtil;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import java.io.File;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class ConsumeQueueStore {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
protected final MessageStore messageStore;
protected final MessageStoreConfig messageStoreConfig;
protected final QueueOffsetAssigner queueOffsetAssigner = new QueueOffsetAssigner();
protected final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface>> consumeQueueTable;
public ConsumeQueueStore(MessageStore messageStore, MessageStoreConfig messageStoreConfig, ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> consumeQueueTable) {
// Should be careful, do not change the topic config
// TopicConfigManager is more suitable here.
private ConcurrentMap<String, TopicConfig> topicConfigTable;
public ConsumeQueueStore(MessageStore messageStore, MessageStoreConfig messageStoreConfig) {
this.messageStore = messageStore;
this.messageStoreConfig = messageStoreConfig;
this.consumeQueueTable = consumeQueueTable;
this.consumeQueueTable = new ConcurrentHashMap<>(32);
}
public void setTopicConfigTable(ConcurrentMap<String, TopicConfig> topicConfigTable) {
this.topicConfigTable = topicConfigTable;
}
private FileQueueLifeCycle getLifeCycle(String topic, int queueId) {
@@ -64,21 +86,143 @@ public class ConsumeQueueStore {
fileQueueLifeCycle.putMessagePositionInfoWrapper(request);
}
public void putMessagePositionInfoWrapper(DispatchRequest dispatchRequest) {
ConsumeQueueInterface cq = this.findOrCreateConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
this.putMessagePositionInfoWrapper(cq, dispatchRequest);
}
public boolean load(ConsumeQueueInterface consumeQueue) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
return fileQueueLifeCycle.load();
}
public boolean load() {
return loadConsumeQueues() && loadBatchConsumeQueues();
}
private boolean loadBatchConsumeQueues() {
File dirLogic = new File(StorePathConfigHelper.getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
File[] fileTopicList = dirLogic.listFiles();
if (fileTopicList != null) {
for (File fileTopic : fileTopicList) {
String topic = fileTopic.getName();
File[] fileQueueIdList = fileTopic.listFiles();
if (fileQueueIdList != null) {
for (File fileQueueId : fileQueueIdList) {
int queueId;
try {
queueId = Integer.parseInt(fileQueueId.getName());
} catch (NumberFormatException e) {
continue;
}
TopicConfig topicConfig = this.topicConfigTable == null ? null : this.topicConfigTable.get(topic);
// For batch consume queue, the topic config must exist
if (topicConfig == null) {
log.warn("topic: {} has no topic config.", topic);
continue;
}
if (!Objects.equals(CQType.BatchCQ, QueueTypeUtils.getCQType(Optional.of(topicConfig)))) {
log.error("[BUG]topic: {} should be BCQ.", topic);
}
ConsumeQueueInterface logic = new BatchConsumeQueue(
topic,
queueId,
StorePathConfigHelper.getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
this.messageStoreConfig.getMappedFileSizeConsumeQueue(),
this.messageStore);
this.putConsumeQueue(topic, queueId, logic);
if (!this.load(logic)) {
return false;
}
}
}
}
}
log.info("load batch consume queue all over, OK");
return true;
}
private boolean loadConsumeQueues() {
File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
File[] fileTopicList = dirLogic.listFiles();
if (fileTopicList != null) {
for (File fileTopic : fileTopicList) {
String topic = fileTopic.getName();
File[] fileQueueIdList = fileTopic.listFiles();
if (fileQueueIdList != null) {
for (File fileQueueId : fileQueueIdList) {
int queueId;
try {
queueId = Integer.parseInt(fileQueueId.getName());
} catch (NumberFormatException e) {
continue;
}
ConsumeQueueInterface logic = new ConsumeQueue(
topic,
queueId,
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
this.messageStoreConfig.getMappedFileSizeConsumeQueue(),
this.messageStore);
this.putConsumeQueue(topic, queueId, logic);
if (!this.load(logic)) {
return false;
}
}
}
}
}
log.info("load logics queue all over, OK");
return true;
}
public void recover(ConsumeQueueInterface consumeQueue) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
fileQueueLifeCycle.recover();
}
public long recover() {
long maxPhysicOffset = -1;
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.recover(logic);
if (logic.getMaxPhysicOffset() > maxPhysicOffset) {
maxPhysicOffset = logic.getMaxPhysicOffset();
}
}
}
return maxPhysicOffset;
}
public void checkSelf(ConsumeQueueInterface consumeQueue) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
fileQueueLifeCycle.checkSelf();
}
public void checkSelf() {
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
Iterator<Map.Entry<Integer, ConsumeQueueInterface>> itNext = next.getValue().entrySet().iterator();
while (itNext.hasNext()) {
Map.Entry<Integer, ConsumeQueueInterface> cq = itNext.next();
this.checkSelf(cq.getValue());
}
}
}
public boolean flush(ConsumeQueueInterface consumeQueue, int flushLeastPages) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
return fileQueueLifeCycle.flush(flushLeastPages);
@@ -89,9 +233,9 @@ public class ConsumeQueueStore {
fileQueueLifeCycle.destroy();
}
public int deleteExpiredFile(ConsumeQueueInterface consumeQueue, long minOffset) {
public int deleteExpiredFile(ConsumeQueueInterface consumeQueue, long minCommitLogPos) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
return fileQueueLifeCycle.deleteExpiredFile(minOffset);
return fileQueueLifeCycle.deleteExpiredFile(minCommitLogPos);
}
public void truncateDirtyLogicFiles(ConsumeQueueInterface consumeQueue, long phyOffset) {
@@ -124,10 +268,9 @@ public class ConsumeQueueStore {
}
private ConsumeQueueInterface doFindOrCreateConsumeQueue(String topic, int queueId) {
ConcurrentMap<Integer, ConsumeQueueInterface> map = consumeQueueTable.get(topic);
if (null == map) {
ConcurrentMap<Integer, ConsumeQueueInterface> newMap = new ConcurrentHashMap<Integer, ConsumeQueueInterface>(128);
ConcurrentMap<Integer, ConsumeQueueInterface> newMap = new ConcurrentHashMap<>(128);
ConcurrentMap<Integer, ConsumeQueueInterface> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
if (oldMap != null) {
map = oldMap;
@@ -143,7 +286,9 @@ public class ConsumeQueueStore {
ConsumeQueueInterface newLogic;
if (StoreUtil.isStreamMode(this.messageStore)) {
Optional<TopicConfig> topicConfig = this.getTopicConfig(topic);
// TODO maybe the topic has been deleted.
if (Objects.equals(CQType.BatchCQ, QueueTypeUtils.getCQType(topicConfig))) {
newLogic = new BatchConsumeQueue(
topic,
queueId,
@@ -162,7 +307,7 @@ public class ConsumeQueueStore {
queueId,
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
this.messageStoreConfig.getMappedFileSizeConsumeQueue(),
(DefaultMessageStore) this.messageStore);
this.messageStore);
ConsumeQueueInterface oldLogic = map.putIfAbsent(queueId, newLogic);
if (oldLogic != null) {
logic = oldLogic;
@@ -173,4 +318,129 @@ public class ConsumeQueueStore {
return logic;
}
public Long getMaxOffset(String topic, int queueId) {
return this.queueOffsetAssigner.getTopicQueueTable().get(topic + "-" + queueId);
}
public void setTopicQueueTable(HashMap<String, Long> topicQueueTable) {
this.queueOffsetAssigner.setTopicQueueTable(topicQueueTable);
}
public void setBatchTopicQueueTable(HashMap<String, Long> batchTopicQueueTable) {
this.queueOffsetAssigner.setBatchTopicQueueTable(batchTopicQueueTable);
}
public void assignQueueOffset(MessageExtBrokerInner msg, short messageNum) {
FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(msg.getTopic(), msg.getQueueId());
fileQueueLifeCycle.assignQueueOffset(this.queueOffsetAssigner, msg, messageNum);
}
public void removeTopicQueueTable(String topic, Integer queueId) {
this.queueOffsetAssigner.remove(topic, queueId);
}
public ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> getConsumeQueueTable() {
return consumeQueueTable;
}
private void putConsumeQueue(final String topic, final int queueId, final ConsumeQueueInterface consumeQueue) {
ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface> map = this.consumeQueueTable.get(topic);
if (null == map) {
map = new ConcurrentHashMap<>();
map.put(queueId, consumeQueue);
this.consumeQueueTable.put(topic, map);
} else {
map.put(queueId, consumeQueue);
}
}
public void recoverOffsetTable(long minPhyOffset) {
HashMap<String, Long> cqOffsetTable = new HashMap<>(1024);
HashMap<String, Long> bcqOffsetTable = new HashMap<>(1024);
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
String key = logic.getTopic() + "-" + logic.getQueueId();
long maxOffsetInQueue = logic.getMaxOffsetInQueue();
if (Objects.equals(CQType.BatchCQ, logic.getCQType())) {
bcqOffsetTable.put(key, maxOffsetInQueue);
} else {
cqOffsetTable.put(key, maxOffsetInQueue);
}
this.correctMinOffset(logic, minPhyOffset);
}
}
this.setTopicQueueTable(cqOffsetTable);
this.setBatchTopicQueueTable(bcqOffsetTable);
}
public void destroy() {
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.destroy(logic);
}
}
}
public void cleanExpired(long minCommitLogOffset) {
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
String topic = next.getKey();
if (!topic.equals(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC)) {
ConcurrentMap<Integer, ConsumeQueueInterface> queueTable = next.getValue();
Iterator<Map.Entry<Integer, ConsumeQueueInterface>> itQT = queueTable.entrySet().iterator();
while (itQT.hasNext()) {
Map.Entry<Integer, ConsumeQueueInterface> nextQT = itQT.next();
long maxCLOffsetInConsumeQueue = nextQT.getValue().getLastOffset();
if (maxCLOffsetInConsumeQueue == -1) {
log.warn("maybe ConsumeQueue was created just now. topic={} queueId={} maxPhysicOffset={} minLogicOffset={}.",
nextQT.getValue().getTopic(),
nextQT.getValue().getQueueId(),
nextQT.getValue().getMaxPhysicOffset(),
nextQT.getValue().getMinLogicOffset());
} else if (maxCLOffsetInConsumeQueue < minCommitLogOffset) {
log.info(
"cleanExpiredConsumerQueue: {} {} consumer queue destroyed, minCommitLogOffset: {} maxCLOffsetInConsumeQueue: {}",
topic,
nextQT.getKey(),
minCommitLogOffset,
maxCLOffsetInConsumeQueue);
removeTopicQueueTable(nextQT.getValue().getTopic(),
nextQT.getValue().getQueueId());
this.destroy(nextQT.getValue());
itQT.remove();
}
}
if (queueTable.isEmpty()) {
log.info("cleanExpiredConsumerQueue: {},topic destroyed", topic);
it.remove();
}
}
}
}
public void truncateDirty(long phyOffset) {
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
for (ConsumeQueueInterface logic : maps.values()) {
this.truncateDirtyLogicFiles(logic, phyOffset);
}
}
}
public Optional<TopicConfig> getTopicConfig(String topic) {
if (this.topicConfigTable == null) {
return Optional.empty();
}
return Optional.ofNullable(this.topicConfigTable.get(topic));
}
}
store/src/main/java/org/apache/rocketmq/store/queue/FileQueueLifeCycle.java
+2
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.Swappable;
public interface FileQueueLifeCycle extends Swappable {
@@ -32,4 +33,5 @@ public interface FileQueueLifeCycle extends Swappable {
boolean isFirstFileExist();
void correctMinOffset(long minCommitLogOffset);
void putMessagePositionInfoWrapper(DispatchRequest request);
void assignQueueOffset(QueueOffsetAssigner queueOffsetAssigner, MessageExtBrokerInner msg, short messageNum);
}
store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetAssigner.java
+60
View File
@@ -0,0 +1,60 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import java.util.HashMap;
/**
* QueueOffsetAssigner is a component for assigning queue.
*
*/
public class QueueOffsetAssigner {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
private HashMap<String, Long> topicQueueTable = new HashMap<>(1024);
private HashMap<String, Long> batchTopicQueueTable = new HashMap<>(1024);
public HashMap<String, Long> getTopicQueueTable() {
return topicQueueTable;
}
public void setTopicQueueTable(HashMap<String, Long> topicQueueTable) {
this.topicQueueTable = topicQueueTable;
}
public HashMap<String, Long> getBatchTopicQueueTable() {
return batchTopicQueueTable;
}
public void setBatchTopicQueueTable(HashMap<String, Long> batchTopicQueueTable) {
this.batchTopicQueueTable = batchTopicQueueTable;
}
public synchronized void remove(String topic, Integer queueId) {
String topicQueueKey = topic + "-" + queueId;
// Beware of thread-safety
this.topicQueueTable.remove(topicQueueKey);
this.batchTopicQueueTable.remove(topicQueueKey);
log.info("removeQueueFromTopicQueueTable OK Topic: {} QueueId: {}", topic, queueId);
}
}
store/src/main/java/org/apache/rocketmq/store/util/QueueTypeUtils.java
-55
View File
@@ -1,55 +0,0 @@
/*
* 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.store.util;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.StreamMessageStore;
import org.apache.rocketmq.store.queue.CQType;
import java.util.Map;
public class QueueTypeUtils {
@Deprecated
public static CQType getCQType(MessageStore messageStore) {
if (messageStore instanceof DefaultMessageStore) {
return CQType.SimpleCQ;
} else if (messageStore instanceof StreamMessageStore) {
return CQType.BatchCQ;
} else {
throw new RuntimeException("new cq type is not supported now.");
}
}
public static CQType getCQType(TopicConfig topicConfig) {
String attributeName = TopicAttributes.QUEUE_TYPE.getName();
Map<String, String> attributes = topicConfig.getAttributes();
if (attributes == null || attributes.size() == 0) {
return CQType.valueOf(TopicAttributes.QUEUE_TYPE.getDefaultValue());
}
if (attributes.containsKey(attributeName)) {
return CQType.valueOf(attributes.get(attributeName));
} else {
return CQType.valueOf(TopicAttributes.QUEUE_TYPE.getDefaultValue());
}
}
}
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreShutDownTest.java
+1 -2
View File
@@ -23,7 +23,7 @@ import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.After;
import org.junit.Before;
@@ -72,7 +72,6 @@ public class DefaultMessageStoreShutDownTest {
messageStoreConfig.setMaxIndexNum(100 * 100);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setHaListenPort(StoreTestBase.nextPort());
messageStoreConfig.setDefaultCQType(CQType.SimpleCQ.name());
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), null, new BrokerConfig());
}
store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+1 -2
View File
@@ -38,7 +38,7 @@ import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -115,7 +115,6 @@ public class DefaultMessageStoreTest {
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setFlushIntervalConsumeQueue(1);
messageStoreConfig.setHaListenPort(StoreTestBase.nextPort());
messageStoreConfig.setDefaultCQType(CQType.SimpleCQ.name());
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MyMessageArrivingListener(), new BrokerConfig());
}
store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java
+219 -55
View File
@@ -17,9 +17,17 @@
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.message.MessageAccessor;
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.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
import org.apache.rocketmq.store.MessageExtBrokerInner;
@@ -27,21 +35,168 @@ import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Queue;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
public class BatchConsumeMessageTest extends QueueTestBase {
private MessageStore messageStore;
@Before
public void init() throws Exception {
messageStore = createMessageStore(null, true);
messageStore.load();
messageStore.start();
}
@After
public void destroy() {
messageStore.shutdown();
messageStore.destroy();
File file = new File(messageStore.getMessageStoreConfig().getStorePathRootDir());
UtilAll.deleteFile(file);
}
@Test
public void testSendMessagesToCqTopic() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.SimpleCQ, messageStore);
int batchNum = 10;
// case 1 has PROPERTY_INNER_NUM but has no INNER_BATCH_FLAG
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, batchNum);
messageExtBrokerInner.setSysFlag(0);
PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
// case 2 has PROPERTY_INNER_NUM and has INNER_BATCH_FLAG, but is not a batchCq
messageExtBrokerInner = buildMessage(topic, 1);
putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
// case 3 has neither PROPERTY_INNER_NUM nor INNER_BATCH_FLAG.
messageExtBrokerInner = buildMessage(topic, -1);
putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
}
@Test
public void testSendMessagesToBcqTopic() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.BatchCQ, messageStore);
// case 1 has PROPERTY_INNER_NUM but has no INNER_BATCH_FLAG
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, 1);
PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
// case 2 has neither PROPERTY_INNER_NUM nor INNER_BATCH_FLAG.
messageExtBrokerInner = buildMessage(topic, -1);
putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
// case 3 has INNER_BATCH_FLAG but has no PROPERTY_INNER_NUM.
messageExtBrokerInner = buildMessage(topic, 1);
MessageAccessor.clearProperty(messageExtBrokerInner, MessageConst.PROPERTY_INNER_NUM);
messageExtBrokerInner.setSysFlag(MessageSysFlag.INNER_BATCH_FLAG);
putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
}
@Test
public void testConsumeBatchMessage() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.BatchCQ, messageStore);
int batchNum = 10;
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, batchNum);
PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
List<GetMessageResult> results = new ArrayList<>();
for (int i = 0; i < batchNum; i++) {
GetMessageResult result = messageStore.getMessage("whatever", topic, 0, i, Integer.MAX_VALUE, Integer.MAX_VALUE, null);
try {
Assert.assertEquals(GetMessageStatus.FOUND, result.getStatus());
results.add(result);
} finally {
result.release();
}
}
for (GetMessageResult result : results) {
Assert.assertEquals(0, result.getMinOffset());
Assert.assertEquals(batchNum, result.getMaxOffset());
}
}
@Test
public void testNextBeginOffsetConsumeBatchMessage() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.BatchCQ, messageStore);
Random random = new Random();
int putMessageCount = 1000;
Queue<Integer> queue = new ArrayDeque<>();
for (int i = 0; i < putMessageCount; i++) {
int batchNum = random.nextInt(1000) + 2;
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, batchNum);
PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
queue.add(batchNum);
}
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
long pullOffset = 0L;
int getMessageCount = 0;
while (true) {
GetMessageResult getMessageResult = messageStore.getMessage("group", topic, 0, pullOffset, 1, null);
if (Objects.equals(getMessageResult.getStatus(), GetMessageStatus.OFFSET_OVERFLOW_ONE)) {
break;
}
Assert.assertEquals(1, getMessageResult.getMessageQueueOffset().size());
Long baseOffset = getMessageResult.getMessageQueueOffset().get(0);
Integer batchNum = queue.poll();
Assert.assertNotNull(batchNum);
Assert.assertEquals(baseOffset + batchNum, getMessageResult.getNextBeginOffset());
pullOffset = getMessageResult.getNextBeginOffset();
getMessageCount++;
}
Assert.assertEquals(putMessageCount, getMessageCount);
}
@Test
public void testGetOffsetInQueueByTime() throws Exception {
MessageStore messageStore = createMessageStore(null, true, CQType.BatchCQ);
messageStore.load();
messageStore.start();
String topic = "testGetOffsetInQueueByTime";
//The initial min max offset, before and after the creation of consume queue
createTopic(topic, CQType.BatchCQ, messageStore);
Assert.assertTrue(QueueTypeUtils.isBatchCq(messageStore.getTopicConfig(topic)));
// The initial min max offset, before and after the creation of consume queue
Assert.assertEquals(0, messageStore.getMaxOffsetInQueue(topic, 0));
Assert.assertEquals(-1, messageStore.getMinOffsetInQueue(topic, 0));
@@ -54,34 +209,30 @@ public class BatchConsumeMessageTest extends QueueTestBase {
if (i == 7)
timeMid = System.currentTimeMillis();
}
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
Assert.assertEquals(80, messageStore.getOffsetInQueueByTime(topic, 0, timeMid));
Assert.assertEquals(0, messageStore.getMinOffsetInQueue(topic, 0));
Assert.assertEquals(190, messageStore.getMaxOffsetInQueue(topic, 0));
Thread.sleep(5 * 1000);
int maxBatchDeleteFilesNum = messageStore.getMessageStoreConfig().getMaxBatchDeleteFilesNum();
messageStore.getCommitLog().deleteExpiredFile(1L, 100, 12000, true, maxBatchDeleteFilesNum);
Assert.assertEquals(80, messageStore.getOffsetInQueueByTime(topic, 0, timeMid));
Thread.sleep(70 * 1000);
Assert.assertEquals(180, messageStore.getOffsetInQueueByTime(topic, 0, timeMid));
}
@Test
public void testDispatchNormalConsumeQueue() throws Exception {
MessageStore messageStore = createMessageStore(null, true, CQType.SimpleCQ);
messageStore.load();
messageStore.start();
String topic = "TestDispatchBuildConsumeQueue";
int batchNum = 10;
createTopic(topic, CQType.SimpleCQ, messageStore);
long timeStart = System.currentTimeMillis();
long timeMid = -1;
for (int i = 0; i < 100; i++) {
// MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, batchNum);
// messageExtBrokerInner.setSysFlag(0);
// PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
// Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
//
// messageExtBrokerInner = buildMessage(topic, 1);
// putMessageResult = messageStore.putMessage(messageExtBrokerInner);
// Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
for (int i = 0; i < 100; i++) {
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, -1);
PutMessageResult putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
@@ -90,7 +241,9 @@ public class BatchConsumeMessageTest extends QueueTestBase {
if (i == 49)
timeMid = System.currentTimeMillis();
}
Thread.sleep(500);
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
ConsumeQueueInterface consumeQueue = messageStore.getConsumeQueue(topic, 0);
Assert.assertEquals(CQType.SimpleCQ, consumeQueue.getCQType());
//check the consume queue
@@ -106,43 +259,41 @@ public class BatchConsumeMessageTest extends QueueTestBase {
for (int i = -100; i < 100; i += 20) {
Assert.assertEquals(consumeQueue.getOffsetInQueueByTime(timeMid + i), messageStore.getOffsetInQueueByTime(topic, 0, timeMid + i));
}
//check the message time
long latencyAllowed = 20;
long earlistMessageTime = messageStore.getEarliestMessageTime(topic, 0);
Assert.assertTrue(earlistMessageTime > timeStart - 10);
Assert.assertTrue(earlistMessageTime < timeStart + 10);
Assert.assertTrue(earlistMessageTime > timeStart - latencyAllowed);
Assert.assertTrue(earlistMessageTime < timeStart + latencyAllowed);
long messageStoreTime = messageStore.getMessageStoreTimeStamp(topic, 0, 50);
Assert.assertTrue(messageStoreTime > timeMid - 10);
Assert.assertTrue(messageStoreTime < timeMid + 10);
Assert.assertTrue(messageStoreTime > timeMid - latencyAllowed);
Assert.assertTrue(messageStoreTime < timeMid + latencyAllowed);
long commitLogOffset = messageStore.getCommitLogOffsetInQueue(topic, 0, 50);
Assert.assertTrue(commitLogOffset >= messageStore.getMinPhyOffset());
Assert.assertTrue(commitLogOffset <= messageStore.getMaxPhyOffset());
Assert.assertFalse(messageStore.checkInDiskByConsumeOffset(topic, 0, 50));
messageStore.shutdown();
messageStore.destroy();
}
@Test
public void testDispatchBuildBatchConsumeQueue() throws Exception {
MessageStore messageStore = createMessageStore(null, true, CQType.BatchCQ);
messageStore.load();
messageStore.start();
String topic = "testDispatchBuildBatchConsumeQueue";
int batchNum = 10;
long timeStart = System.currentTimeMillis();
long timeMid = -1;
createTopic(topic, CQType.BatchCQ, messageStore);
for (int i = 0; i < 100; i++) {
PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(topic, batchNum));
Assert.assertEquals(PutMessageStatus.PUT_OK, putMessageResult.getPutMessageStatus());
Thread.sleep(2);
if (i == 29)
timeMid = System.currentTimeMillis();
MessageExtBrokerInner messageExtBrokerInner = buildMessage(topic, 1);
putMessageResult = messageStore.putMessage(messageExtBrokerInner);
Assert.assertEquals(PutMessageStatus.MESSAGE_ILLEGAL, putMessageResult.getPutMessageStatus());
}
Thread.sleep(500);
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
ConsumeQueueInterface consumeQueue = messageStore.getConsumeQueue(topic, 0);
Assert.assertEquals(CQType.BatchCQ, consumeQueue.getCQType());
@@ -176,21 +327,18 @@ public class BatchConsumeMessageTest extends QueueTestBase {
for (int i = 0; i < 10; i++) {
SelectMappedBufferResult sbr = getMessageResult.getMessageMapedList().get(i);
MessageExt messageExt = MessageDecoder.decode(sbr.getByteBuffer());
short tmpBatchNum = Short.valueOf(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
short tmpBatchNum = Short.parseShort(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
Assert.assertEquals(i * batchNum, Long.parseLong(messageExt.getProperty(MessageConst.PROPERTY_INNER_BASE)));
Assert.assertEquals(batchNum, tmpBatchNum);
}
messageStore.destroy();
messageStore.shutdown();
}
@Test
public void testGetBatchMessageWithinNumber() throws Exception {
MessageStore messageStore = createMessageStore(null, true, CQType.BatchCQ);
messageStore.load();
messageStore.start();
public void testGetBatchMessageWithinNumber() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.BatchCQ, messageStore);
int batchNum = 20;
for (int i = 0; i < 200; i++) {
PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(topic, batchNum));
@@ -198,7 +346,9 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertEquals(i * batchNum, putMessageResult.getAppendMessageResult().getLogicsOffset());
Assert.assertEquals(batchNum, putMessageResult.getAppendMessageResult().getMsgNum());
}
Thread.sleep(500);
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
ConsumeQueueInterface consumeQueue = messageStore.getConsumeQueue(topic, 0);
Assert.assertEquals(CQType.BatchCQ, consumeQueue.getCQType());
Assert.assertEquals(0, consumeQueue.getMinOffsetInQueue());
@@ -212,7 +362,7 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertEquals(batchNum, getMessageResult.getMessageCount());
SelectMappedBufferResult sbr = getMessageResult.getMessageMapedList().get(0);
MessageExt messageExt = MessageDecoder.decode(sbr.getByteBuffer());
short tmpBatchNum = Short.valueOf(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
short tmpBatchNum = Short.parseShort(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
Assert.assertEquals(0, messageExt.getQueueOffset());
Assert.assertEquals(batchNum, tmpBatchNum);
}
@@ -236,22 +386,19 @@ public class BatchConsumeMessageTest extends QueueTestBase {
SelectMappedBufferResult sbr = getMessageResult.getMessageMapedList().get(i);
MessageExt messageExt = MessageDecoder.decode(sbr.getByteBuffer());
Assert.assertNotNull(messageExt);
short innerBatchNum = Short.valueOf(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
short innerBatchNum = Short.parseShort(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
Assert.assertEquals(i * batchNum, Long.parseLong(messageExt.getProperty(MessageConst.PROPERTY_INNER_BASE)));
Assert.assertEquals(batchNum, innerBatchNum);
}
}
messageStore.destroy();
messageStore.shutdown();
}
@Test
public void testGetBatchMessageWithinSize() throws Exception {
MessageStore messageStore = createMessageStore(null, true, CQType.BatchCQ);
messageStore.load();
messageStore.start();
public void testGetBatchMessageWithinSize() {
String topic = UUID.randomUUID().toString();
createTopic(topic, CQType.BatchCQ, messageStore);
int batchNum = 10;
for (int i = 0; i < 100; i++) {
PutMessageResult putMessageResult = messageStore.putMessage(buildMessage(topic, batchNum));
@@ -259,7 +406,9 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertEquals(i * 10, putMessageResult.getAppendMessageResult().getLogicsOffset());
Assert.assertEquals(batchNum, putMessageResult.getAppendMessageResult().getMsgNum());
}
Thread.sleep(500);
await().atMost(5, SECONDS).until(fullyDispatched(messageStore));
ConsumeQueueInterface consumeQueue = messageStore.getConsumeQueue(topic, 0);
Assert.assertEquals(CQType.BatchCQ, consumeQueue.getCQType());
Assert.assertEquals(0, consumeQueue.getMinOffsetInQueue());
@@ -272,7 +421,7 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertEquals(1, getMessageResult.getMessageMapedList().size());
SelectMappedBufferResult sbr = getMessageResult.getMessageMapedList().get(0);
MessageExt messageExt = MessageDecoder.decode(sbr.getByteBuffer());
short tmpBatchNum = Short.valueOf(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
short tmpBatchNum = Short.parseShort(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
Assert.assertEquals(0, messageExt.getQueueOffset());
Assert.assertEquals(batchNum, tmpBatchNum);
}
@@ -293,14 +442,29 @@ public class BatchConsumeMessageTest extends QueueTestBase {
Assert.assertFalse(getMessageResult.getMessageMapedList().get(i).hasReleased());
SelectMappedBufferResult sbr = getMessageResult.getMessageMapedList().get(i);
MessageExt messageExt = MessageDecoder.decode(sbr.getByteBuffer());
short tmpBatchNum = Short.valueOf(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
short tmpBatchNum = Short.parseShort(messageExt.getProperty(MessageConst.PROPERTY_INNER_NUM));
Assert.assertEquals(i * batchNum, Long.parseLong(messageExt.getProperty(MessageConst.PROPERTY_INNER_BASE)));
Assert.assertEquals(batchNum, tmpBatchNum);
}
}
messageStore.destroy();
messageStore.shutdown();
}
private void createTopic(String topic, CQType cqType, MessageStore messageStore) {
ConcurrentMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
TopicConfig topicConfigToBeAdded = new TopicConfig();
Map<String, String> attributes = new HashMap<>();
attributes.put(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName(), cqType.toString());
topicConfigToBeAdded.setTopicName(topic);
topicConfigToBeAdded.setAttributes(attributes);
topicConfigTable.put(topic, topicConfigToBeAdded);
((DefaultMessageStore)messageStore).setTopicConfigTable(topicConfigTable);
}
private Callable<Boolean> fullyDispatched(MessageStore messageStore) {
return () -> messageStore.dispatchBehindBytes() == 0;
}
}
store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java
+11 -16
View File
@@ -19,8 +19,8 @@ package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.StreamMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.StoreTestBase;
import org.apache.rocketmq.store.config.MessageStoreConfig;
@@ -32,7 +32,6 @@ import org.junit.Test;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import static java.lang.String.format;
@@ -45,13 +44,13 @@ public class BatchConsumeQueueTest extends StoreTestBase {
path = createBaseDir();
}
baseDirs.add(path);
StreamMessageStore bcqMessageStore = null;
MessageStore messageStore = null;
try {
bcqMessageStore = createBcqMessageStore(null);
messageStore = createMessageStore(null);
} catch (Exception e) {
Assert.fail();
}
BatchConsumeQueue batchConsumeQueue = new BatchConsumeQueue("topic", 0, path,fileSize, bcqMessageStore);
BatchConsumeQueue batchConsumeQueue = new BatchConsumeQueue("topic", 0, path,fileSize, messageStore);
batchConsumeQueues.add(batchConsumeQueue);
return batchConsumeQueue;
}
@@ -280,7 +279,7 @@ public class BatchConsumeQueueTest extends StoreTestBase {
}
}
protected StreamMessageStore createBcqMessageStore(String baseDir) throws Exception {
protected MessageStore createMessageStore(String baseDir) throws Exception {
if (baseDir == null) {
baseDir = createBaseDir();
}
@@ -299,17 +298,13 @@ public class BatchConsumeQueueTest extends StoreTestBase {
messageStoreConfig.setMaxTransferBytesOnMessageInMemory(1024 * 1024);
messageStoreConfig.setMaxTransferCountOnMessageInDisk(1024);
messageStoreConfig.setMaxTransferCountOnMessageInMemory(1024);
messageStoreConfig.setDefaultCQType(CQType.BatchCQ.name());
messageStoreConfig.setSearchBcqByCacheEnable(true);
StreamMessageStore defaultMessageStore = new StreamMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MessageArrivingListener() {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
byte[] filterBitMap, Map<String, String> properties) {
}
}, new BrokerConfig());
return defaultMessageStore;
return new DefaultMessageStore(
messageStoreConfig,
new BrokerStatsManager("simpleTest"),
(topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties) -> {},
new BrokerConfig());
}
}
store/src/test/java/org/apache/rocketmq/store/queue/ConsumeQueueTest.java
+2 -1
View File
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.store.queue;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.store.ConsumeQueueExt;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.MessageStore;
@@ -30,7 +31,7 @@ public class ConsumeQueueTest extends QueueTestBase {
public void testIterator() throws Exception {
final int msgNum = 100;
final int msgSize = 1000;
MessageStore messageStore = createMessageStore(null, true, CQType.SimpleCQ);
MessageStore messageStore = createMessageStore(null, true);
messageStore.load();
String topic = UUID.randomUUID().toString();
//The initial min max offset, before and after the creation of consume queue
store/src/test/java/org/apache/rocketmq/store/queue/QueueTestBase.java
+8 -20
View File
@@ -23,7 +23,6 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.StreamMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
@@ -33,11 +32,10 @@ import org.apache.rocketmq.store.stats.BrokerStatsManager;
import java.io.File;
import java.util.Map;
import java.util.Objects;
public class QueueTestBase extends StoreTestBase {
protected MessageStore createMessageStore(String baseDir, boolean extent, CQType cqType) throws Exception {
protected MessageStore createMessageStore(String baseDir, boolean extent) throws Exception {
if (baseDir == null) {
baseDir = createBaseDir();
}
@@ -56,28 +54,18 @@ public class QueueTestBase extends StoreTestBase {
messageStoreConfig.setMaxTransferBytesOnMessageInMemory(1024 * 1024);
messageStoreConfig.setMaxTransferCountOnMessageInDisk(1024);
messageStoreConfig.setMaxTransferCountOnMessageInMemory(1024);
messageStoreConfig.setDefaultCQType(cqType.name());
messageStoreConfig.setFlushIntervalCommitLog(1);
messageStoreConfig.setFlushCommitLogThoroughInterval(2);
if (Objects.equals(CQType.BatchCQ, cqType)) {
return new StreamMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MessageArrivingListener() {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
byte[] filterBitMap, Map<String, String> properties) {
DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MessageArrivingListener() {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
byte[] filterBitMap, Map<String, String> properties) {
}
}, new BrokerConfig());
} else {
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MessageArrivingListener() {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
byte[] filterBitMap, Map<String, String> properties) {
}
}, new BrokerConfig());
}
}
}, new BrokerConfig());
return messageStore;
}
public MessageExtBrokerInner buildMessage(String topic, int batchNum) {
test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
+4 -4
View File
@@ -50,20 +50,20 @@ public class MQAdminTestUtils {
private static Logger log = Logger.getLogger(MQAdminTestUtils.class);
public static boolean createTopic(String nameSrvAddr, String clusterName, String topic,
int queueNum) {
int queueNum, Map<String, String> attributes) {
int defaultWaitTime = 5;
return createTopic(nameSrvAddr, clusterName, topic, queueNum, defaultWaitTime);
return createTopic(nameSrvAddr, clusterName, topic, queueNum, attributes, defaultWaitTime);
}
public static boolean createTopic(String nameSrvAddr, String clusterName, String topic,
int queueNum, int waitTimeSec) {
int queueNum, Map<String, String> attributes, int waitTimeSec) {
boolean createResult = false;
DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
mqAdminExt.setInstanceName(UUID.randomUUID().toString());
mqAdminExt.setNamesrvAddr(nameSrvAddr);
try {
mqAdminExt.start();
mqAdminExt.createTopic(clusterName, topic, queueNum);
mqAdminExt.createTopic(clusterName, topic, queueNum, attributes);
} catch (Exception e) {
}
test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
+14 -21
View File
@@ -37,7 +37,7 @@ import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.test.client.rmq.RMQAsyncSendProducer;
import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
@@ -61,18 +61,14 @@ public class BaseConf {
//the logic queue test need at least three brokers
protected final static String broker3Name;
protected final static String clusterName;
protected final static String steamClusterName = "steam-cluster";
protected final static int brokerNum;
protected final static int waitTime = 5;
protected final static int consumeTime = 2 * 60 * 1000;
protected final static int QUEUE_NUMBERS = 8;
protected final static NamesrvController namesrvController;
protected static BrokerController brokerController1;
protected static BrokerController brokerController2;
protected static BrokerController brokerController3;
protected static BrokerController streamBrokerController1;
protected static BrokerController streamBrokerController2;
protected static BrokerController streamBrokerController3;
protected final static BrokerController brokerController1;
protected final static BrokerController brokerController2;
protected final static BrokerController brokerController3;
protected final static List<BrokerController> brokerControllerList;
protected final static Map<String, BrokerController> brokerControllerMap;
protected final static List<Object> mqClients = new ArrayList<Object>();
@@ -80,7 +76,7 @@ public class BaseConf {
private final static Logger log = Logger.getLogger(BaseConf.class);
static {
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
namesrvController = IntegrationTestBase.createAndStartNamesrv();
nsAddr = "127.0.0.1:" + namesrvController.getNettyServerConfig().getListenPort();
brokerController1 = IntegrationTestBase.createAndStartBroker(nsAddr);
@@ -91,9 +87,6 @@ public class BaseConf {
broker2Name = brokerController2.getBrokerConfig().getBrokerName();
broker3Name = brokerController3.getBrokerConfig().getBrokerName();
brokerNum = 3;
streamBrokerController1 = IntegrationTestBase.createAndStartBroker(nsAddr, CQType.BatchCQ.toString(), steamClusterName);
streamBrokerController2 = IntegrationTestBase.createAndStartBroker(nsAddr, CQType.BatchCQ.toString(), steamClusterName);
streamBrokerController3 = IntegrationTestBase.createAndStartBroker(nsAddr, CQType.BatchCQ.toString(), steamClusterName);
brokerControllerList = ImmutableList.of(brokerController1, brokerController2, brokerController3);
brokerControllerMap = brokerControllerList.stream().collect(Collectors.toMap(input -> input.getBrokerConfig().getBrokerName(), Function.identity()));
}
@@ -124,7 +117,7 @@ public class BaseConf {
public static String initTopic() {
String topic = "tt-" + MQRandomUtils.getRandomTopic();
IntegrationTestBase.initTopic(topic, nsAddr, clusterName);
IntegrationTestBase.initTopic(topic, nsAddr, clusterName, CQType.SimpleCQ);
return topic;
}
@@ -171,9 +164,9 @@ public class BaseConf {
}
public static RMQNormalProducer getProducer(String nsAddr, String topic, String producerGoup,
String instanceName) {
String instanceName) {
RMQNormalProducer producer = new RMQNormalProducer(nsAddr, topic, producerGoup,
instanceName);
instanceName);
if (debug) {
producer.setDebug();
}
@@ -191,31 +184,31 @@ public class BaseConf {
}
public static RMQNormalConsumer getConsumer(String nsAddr, String topic, String subExpression,
AbstractListener listener) {
AbstractListener listener) {
return getConsumer(nsAddr, topic, subExpression, listener, false);
}
public static RMQNormalConsumer getConsumer(String nsAddr, String topic, String subExpression,
AbstractListener listener, boolean useTLS) {
AbstractListener listener, boolean useTLS) {
String consumerGroup = initConsumerGroup();
return getConsumer(nsAddr, consumerGroup, topic, subExpression, listener, useTLS);
}
public static RMQNormalConsumer getConsumer(String nsAddr, String consumerGroup, String topic,
String subExpression, AbstractListener listener) {
String subExpression, AbstractListener listener) {
return getConsumer(nsAddr, consumerGroup, topic, subExpression, listener, false);
}
public static RMQNormalConsumer getConsumer(String nsAddr, String consumerGroup, String topic,
String subExpression, AbstractListener listener, boolean useTLS) {
String subExpression, AbstractListener listener, boolean useTLS) {
RMQNormalConsumer consumer = ConsumerFactory.getRMQNormalConsumer(nsAddr, consumerGroup,
topic, subExpression, listener, useTLS);
topic, subExpression, listener, useTLS);
if (debug) {
consumer.setDebug();
}
mqClients.add(consumer);
log.info(String.format("consumer[%s] start,topic[%s],subExpression[%s]", consumerGroup,
topic, subExpression));
topic, subExpression));
return consumer;
}
test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
+14 -35
View File
@@ -20,13 +20,18 @@ package org.apache.rocketmq.test.base;
import com.google.common.truth.Truth;
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -34,10 +39,8 @@ import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.test.util.MQAdminTestUtils;
import org.apache.rocketmq.test.util.TestUtils;
import org.assertj.core.util.Strings;
public class IntegrationTestBase {
public static InternalLogger logger = InternalLoggerFactory.getLogger(IntegrationTestBase.class);
@@ -142,34 +145,6 @@ public class IntegrationTestBase {
storeConfig.setMaxIndexNum(INDEX_NUM);
storeConfig.setMaxHashSlotNum(INDEX_NUM * 4);
storeConfig.setDeleteWhen("01;02;03;04;05;06;07;08;09;10;11;12;13;14;15;16;17;18;19;20;21;22;23;00");
storeConfig.setDefaultCQType(CQType.SimpleCQ.toString());
storeConfig.setMaxTransferCountOnMessageInMemory(1024);
storeConfig.setMaxTransferCountOnMessageInDisk(1024);
return createAndStartBroker(storeConfig, brokerConfig);
}
public static BrokerController createAndStartBroker(String nsAddr, String cqType, String cluster) {
String baseDir = createBaseDir();
BrokerConfig brokerConfig = new BrokerConfig();
MessageStoreConfig storeConfig = new MessageStoreConfig();
brokerConfig.setBrokerName(BROKER_NAME_PREFIX + BROKER_INDEX.incrementAndGet());
brokerConfig.setBrokerIP1("127.0.0.1");
brokerConfig.setNamesrvAddr(nsAddr);
brokerConfig.setEnablePropertyFilter(true);
brokerConfig.setLoadBalancePollNameServerInterval(500);
brokerConfig.setBrokerClusterName(cluster);
storeConfig.setStorePathRootDir(baseDir);
storeConfig.setStorePathCommitLog(baseDir + SEP + "commitlog");
storeConfig.setMappedFileSizeCommitLog(COMMIT_LOG_SIZE);
storeConfig.setMaxIndexNum(INDEX_NUM);
storeConfig.setMaxHashSlotNum(INDEX_NUM * 4);
storeConfig.setDeleteWhen("01;02;03;04;05;06;07;08;09;10;11;12;13;14;15;16;17;18;19;20;21;22;23;00");
if (!Strings.isNullOrEmpty(cqType)) {
storeConfig.setDefaultCQType(cqType);
} else {
storeConfig.setDefaultCQType(CQType.SimpleCQ.toString());
}
storeConfig.setMaxTransferCountOnMessageInMemory(1024);
storeConfig.setMaxTransferCountOnMessageInDisk(1024);
return createAndStartBroker(storeConfig, brokerConfig);
@@ -193,17 +168,21 @@ public class IntegrationTestBase {
return brokerController;
}
public static boolean initTopic(String topic, String nsAddr, String clusterName, int queueNumbers) {
public static boolean initTopic(String topic, String nsAddr, String clusterName, int queueNumbers, CQType cqType) {
long startTime = System.currentTimeMillis();
boolean createResult;
while (true) {
createResult = MQAdminTestUtils.createTopic(nsAddr, clusterName, topic, queueNumbers);
Map<String, String> attributes = new HashMap<>();
if (!Objects.equals(CQType.SimpleCQ, cqType)) {
attributes.put("+" + TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName(), cqType.toString());
}
createResult = MQAdminTestUtils.createTopic(nsAddr, clusterName, topic, queueNumbers, attributes);
if (createResult) {
break;
} else if (System.currentTimeMillis() - startTime > topicCreateTime) {
Truth.assertWithMessage(String.format("topic[%s] is created failed after:%d ms", topic,
System.currentTimeMillis() - startTime)).fail();
System.currentTimeMillis() - startTime)).fail();
break;
} else {
TestUtils.waitForMoment(500);
@@ -214,8 +193,8 @@ public class IntegrationTestBase {
return createResult;
}
public static boolean initTopic(String topic, String nsAddr, String clusterName) {
return initTopic(topic, nsAddr, clusterName, BaseConf.QUEUE_NUMBERS);
public static boolean initTopic(String topic, String nsAddr, String clusterName, CQType cqType) {
return initTopic(topic, nsAddr, clusterName, BaseConf.QUEUE_NUMBERS, cqType);
}
public static void deleteFile(File file) {
test/src/test/java/org/apache/rocketmq/test/base/dledger/DLedgerProduceAndConsumeIT.java
+2 -3
View File
@@ -30,7 +30,7 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.test.base.IntegrationTestBase;
import org.apache.rocketmq.test.factory.ConsumerFactory;
@@ -63,7 +63,6 @@ public class DLedgerProduceAndConsumeIT {
storeConfig.setdLegerGroup(brokerName);
storeConfig.setdLegerSelfId(selfId);
storeConfig.setdLegerPeers(peers);
storeConfig.setDefaultCQType(CQType.SimpleCQ.toString());
return storeConfig;
}
@@ -83,7 +82,7 @@ public class DLedgerProduceAndConsumeIT {
String topic = UUID.randomUUID().toString();
String consumerGroup = UUID.randomUUID().toString();
IntegrationTestBase.initTopic(topic, BaseConf.nsAddr, cluster, 1);
IntegrationTestBase.initTopic(topic, BaseConf.nsAddr, cluster, 1, CQType.SimpleCQ);
DefaultMQProducer producer = ProducerFactory.getRMQProducer(BaseConf.nsAddr);
DefaultMQPullConsumer consumer = ConsumerFactory.getRMQPullConsumer(BaseConf.nsAddr, consumerGroup);
test/src/test/java/org/apache/rocketmq/test/client/producer/batch/BatchSendIT.java
+18 -19
View File
@@ -23,7 +23,6 @@ import java.util.Random;
import java.util.UUID;
import org.apache.log4j.Logger;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
@@ -32,14 +31,14 @@ import org.apache.rocketmq.client.hook.SendMessageHook;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
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.store.DefaultMessageStore;
import org.apache.rocketmq.store.StreamMessageStore;
import org.apache.rocketmq.store.queue.CQType;
import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.test.base.IntegrationTestBase;
import org.apache.rocketmq.test.client.consumer.tag.TagMessageWith1ConsumerIT;
@@ -101,23 +100,23 @@ public class BatchSendIT extends BaseConf {
@Test
public void testBatchSend_SysInnerBatch() throws Exception {
waitBrokerRegistered(nsAddr, steamClusterName, brokerNum);
Assert.assertTrue(streamBrokerController1.getMessageStore() instanceof StreamMessageStore);
Assert.assertTrue(streamBrokerController2.getMessageStore() instanceof StreamMessageStore);
Assert.assertTrue(streamBrokerController3.getMessageStore() instanceof StreamMessageStore);
waitBrokerRegistered(nsAddr, clusterName, brokerNum);
String batchTopic = UUID.randomUUID().toString();
IntegrationTestBase.initTopic(batchTopic, nsAddr, steamClusterName);
Assert.assertEquals(8, streamBrokerController1.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, streamBrokerController2.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, streamBrokerController3.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(-1, streamBrokerController1.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(-1, streamBrokerController2.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(-1, streamBrokerController3.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, streamBrokerController1.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, streamBrokerController2.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, streamBrokerController3.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
IntegrationTestBase.initTopic(batchTopic, nsAddr, clusterName, CQType.BatchCQ);
Assert.assertEquals(CQType.BatchCQ.toString(), brokerController1.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getAttributes().get(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName()));
Assert.assertEquals(CQType.BatchCQ.toString(), brokerController2.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getAttributes().get(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName()));
Assert.assertEquals(CQType.BatchCQ.toString(), brokerController3.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getAttributes().get(TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName()));
Assert.assertEquals(8, brokerController1.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, brokerController2.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, brokerController3.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(-1, brokerController1.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(-1, brokerController2.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(-1, brokerController3.getMessageStore().getMinOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, brokerController1.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, brokerController2.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
Assert.assertEquals(0, brokerController3.getMessageStore().getMaxOffsetInQueue(batchTopic, 0));
DefaultMQProducer producer = ProducerFactory.getRMQProducer(nsAddr);
MessageQueue messageQueue = producer.fetchPublishMessageQueues(batchTopic).iterator().next();
@@ -168,7 +167,7 @@ public class BatchSendIT extends BaseConf {
Assert.assertTrue(brokerController3.getMessageStore() instanceof DefaultMessageStore);
String batchTopic = UUID.randomUUID().toString();
IntegrationTestBase.initTopic(batchTopic, nsAddr, clusterName);
IntegrationTestBase.initTopic(batchTopic, nsAddr, clusterName, CQType.SimpleCQ);
Assert.assertEquals(8, brokerController1.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, brokerController2.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
Assert.assertEquals(8, brokerController3.getTopicConfigManager().getTopicConfigTable().get(batchTopic).getReadQueueNums());
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+4 -4
View File
@@ -95,13 +95,13 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, newTopic, queueNum, 0);
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
createTopic(key, newTopic, queueNum, 0, attributes);
}
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
defaultMQAdminExtImpl.createTopic(key, newTopic, queueNum, topicSysFlag);
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
defaultMQAdminExtImpl.createTopic(key, newTopic, queueNum, topicSysFlag, attributes);
}
@Override
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+5 -11
View File
@@ -77,11 +77,8 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingOne;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.RPCHook;
@@ -95,9 +92,6 @@ import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.tools.admin.api.MessageTrack;
import org.apache.rocketmq.tools.admin.api.TrackType;
import static org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils.checkAndBuildMappingItems;
import static org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils.getMappingDetailFromConfig;
public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
private static final Set<String> SYSTEM_GROUP_SET = new HashSet<String>();
@@ -1063,13 +1057,13 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
createTopic(key, newTopic, queueNum, 0);
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
createTopic(key, newTopic, queueNum, 0, null);
}
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
this.mqClientInstance.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes) throws MQClientException {
this.mqClientInstance.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag, attributes);
}
@Override
@@ -1176,7 +1170,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public void setMessageRequestMode(final String brokerAddr, final String topic, final String consumerGroup, final
MessageRequestMode mode, final int popShareQueueNum, final long timeoutMillis)
MessageRequestMode mode, final int popShareQueueNum, final long timeoutMillis)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
RemotingConnectException, MQClientException {
this.mqClientInstance.getMQClientAPIImpl().setMessageRequestMode(brokerAddr, topic, consumerGroup, mode, popShareQueueNum, timeoutMillis);
tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java
-1
View File
@@ -23,7 +23,6 @@ import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionGroup;
import org.apache.commons.cli.Options;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.AttributeParser;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;