opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-29 04:13:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #8589] Support file format CQ and json format offset in-place upgrade to rocksdb management (#8600)

Browse Source
This commit is contained in:
LetLetMe
2024-09-23 19:24:15 +08:00
committed by GitHub
parent e8d1472fb7
commit 525f877f3b
25 changed files with 755 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+4 -2
View File
@@ -18,7 +18,6 @@ package org.apache.rocketmq.broker;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
@@ -789,6 +788,9 @@ public class BrokerController {
defaultMessageStore = new RocksDBMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
} else {
defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
if (messageStoreConfig.isRocksdbCQDoubleWriteEnable()) {
defaultMessageStore.enableRocksdbCQWrite();
}
}
if (messageStoreConfig.isEnableDLegerCommitLog()) {
@@ -812,7 +814,7 @@ public class BrokerController {
this.timerMessageStore.registerEscapeBridgeHook(msg -> escapeBridge.putMessage(msg));
this.messageStore.setTimerMessageStore(this.timerMessageStore);
}
} catch (IOException e) {
} catch (Exception e) {
result = false;
LOG.error("BrokerController#initialize: unexpected error occurs", e);
}
broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
+20
View File
@@ -31,6 +31,7 @@ import com.google.common.base.Strings;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerPathConfigHelper;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -373,6 +374,25 @@ public class ConsumerOffsetManager extends ConfigManager {
this.dataVersion = dataVersion;
}
public boolean loadDataVersion() {
String fileName = null;
try {
fileName = this.configFilePath();
String jsonString = MixAll.file2String(fileName);
if (jsonString != null) {
ConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, ConsumerOffsetManager.class);
if (obj != null) {
this.dataVersion = obj.dataVersion;
}
LOG.info("load consumer offset dataVersion success,{},{} ", fileName, jsonString);
}
return true;
} catch (Exception e) {
LOG.error("load consumer offset dataVersion failed " + fileName, e);
return false;
}
}
public void removeOffset(final String group) {
Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
while (it.hasNext()) {
broker/src/main/java/org/apache/rocketmq/broker/offset/RocksDBConsumerOffsetManager.java
+69 -8
View File
@@ -16,26 +16,31 @@
*/
package org.apache.rocketmq.broker.offset;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import java.io.File;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.RocksDBConfigManager;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.DataConverter;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.rocksdb.WriteBatch;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
protected static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
protected RocksDBConfigManager rocksDBConfigManager;
public RocksDBConsumerOffsetManager(BrokerController brokerController) {
super(brokerController);
this.rocksDBConfigManager = new RocksDBConfigManager(configFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs());
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs());
}
@Override
@@ -43,9 +48,47 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
if (!rocksDBConfigManager.init()) {
return false;
}
return this.rocksDBConfigManager.loadData(this::decodeOffset);
if (!loadDataVersion() || !loadConsumerOffset()) {
return false;
}
return true;
}
public boolean loadConsumerOffset() {
return this.rocksDBConfigManager.loadData(this::decodeOffset) && merge();
}
private boolean merge() {
if (!brokerController.getMessageStoreConfig().isTransferOffsetJsonToRocksdb()) {
log.info("the switch transferOffsetJsonToRocksdb is off, no merge offset operation is needed.");
return true;
}
if (!UtilAll.isPathExists(this.configFilePath()) && !UtilAll.isPathExists(this.configFilePath() + ".bak")) {
log.info("consumerOffset json file does not exist, so skip merge");
return true;
}
if (!super.loadDataVersion()) {
log.error("load json consumerOffset dataVersion error, startup will exit");
return false;
}
final DataVersion dataVersion = super.getDataVersion();
final DataVersion kvDataVersion = this.getDataVersion();
if (dataVersion.getCounter().get() > kvDataVersion.getCounter().get()) {
if (!super.load()) {
log.error("load json consumerOffset info failed, startup will exit");
return false;
}
this.persist();
this.getDataVersion().assignNewOne(dataVersion);
updateDataVersion();
log.info("update offset from json, dataVersion:{}, offsetTable: {} ", this.getDataVersion(), JSON.toJSONString(this.getOffsetTable()));
}
return true;
}
@Override
public boolean stop() {
return this.rocksDBConfigManager.stop();
@@ -69,8 +112,7 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
LOG.info("load exist local offset, {}, {}", topicAtGroup, wrapper.getOffsetTable());
}
@Override
public String configFilePath() {
public String rocksdbConfigFilePath() {
return this.brokerController.getMessageStoreConfig().getStorePathRootDir() + File.separator + "config" + File.separator + "consumerOffsets" + File.separator;
}
@@ -103,4 +145,23 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
byte[] valueBytes = JSON.toJSONBytes(wrapper, SerializerFeature.BrowserCompatible);
writeBatch.put(keyBytes, valueBytes);
}
@Override
public boolean loadDataVersion() {
return this.rocksDBConfigManager.loadDataVersion();
}
@Override
public DataVersion getDataVersion() {
return rocksDBConfigManager.getKvDataVersion();
}
public void updateDataVersion() {
try {
rocksDBConfigManager.updateKvDataVersion();
} catch (Exception e) {
log.error("update consumer offset dataVersion error", e);
throw new RuntimeException(e);
}
}
}
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+90 -2
View File
@@ -18,9 +18,11 @@ package org.apache.rocketmq.broker.processor;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.opentelemetry.api.common.Attributes;
import java.io.UnsupportedEncodingException;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
@@ -38,7 +40,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import io.opentelemetry.api.common.Attributes;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.AccessValidator;
@@ -69,6 +70,7 @@ import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.LockCallback;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UnlockCallback;
@@ -137,6 +139,7 @@ import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerialize
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.body.UserInfo;
import org.apache.rocketmq.remoting.protocol.header.CheckRocksdbCqWriteProgressRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.CloneGroupOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.CreateAccessConfigRequestHeader;
@@ -209,6 +212,7 @@ import org.apache.rocketmq.store.MessageFilter;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.RocksDBMessageStore;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
@@ -217,8 +221,9 @@ import org.apache.rocketmq.store.queue.ReferredIterator;
import org.apache.rocketmq.store.timer.TimerCheckpoint;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import org.apache.rocketmq.store.util.LibC;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_SYSTEM;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_INVOCATION_STATUS;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_SYSTEM;
import static org.apache.rocketmq.remoting.protocol.RemotingCommand.buildErrorResponse;
public class AdminBrokerProcessor implements NettyRequestProcessor {
@@ -339,6 +344,8 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
return fetchAllConsumeStatsInBroker(ctx, request);
case RequestCode.QUERY_CONSUME_QUEUE:
return queryConsumeQueue(ctx, request);
case RequestCode.CHECK_ROCKSDB_CQ_WRITE_PROGRESS:
return this.checkRocksdbCqWriteProgress(ctx, request);
case RequestCode.UPDATE_AND_GET_GROUP_FORBIDDEN:
return this.updateAndGetGroupForbidden(ctx, request);
case RequestCode.GET_SUBSCRIPTIONGROUP_CONFIG:
@@ -458,6 +465,71 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
return response;
}
private RemotingCommand checkRocksdbCqWriteProgress(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
CheckRocksdbCqWriteProgressRequestHeader requestHeader = request.decodeCommandCustomHeader(CheckRocksdbCqWriteProgressRequestHeader.class);
String requestTopic = requestHeader.getTopic();
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
response.setCode(ResponseCode.SUCCESS);
DefaultMessageStore messageStore = (DefaultMessageStore) brokerController.getMessageStore();
RocksDBMessageStore rocksDBMessageStore = messageStore.getRocksDBMessageStore();
if (!messageStore.getMessageStoreConfig().isRocksdbCQDoubleWriteEnable()) {
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult", "rocksdbCQWriteEnable is false, checkRocksdbCqWriteProgressCommand is invalid")));
return response;
}
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> cqTable = messageStore.getConsumeQueueTable();
StringBuilder diffResult = new StringBuilder("check success, all is ok!\n");
try {
if (StringUtils.isNotBlank(requestTopic)) {
processConsumeQueuesForTopic(cqTable.get(requestTopic), requestTopic, rocksDBMessageStore, diffResult,false);
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult", diffResult.toString())));
return response;
}
for (Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> topicEntry : cqTable.entrySet()) {
String topic = topicEntry.getKey();
processConsumeQueuesForTopic(topicEntry.getValue(), topic, rocksDBMessageStore, diffResult,true);
}
diffResult.append("check all topic successful, size:").append(cqTable.size());
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult", diffResult.toString())));
} catch (Exception e) {
LOGGER.error("CheckRocksdbCqWriteProgressCommand error", e);
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult", e.getMessage())));
}
return response;
}
private void processConsumeQueuesForTopic(ConcurrentMap<Integer, ConsumeQueueInterface> queueMap, String topic, RocksDBMessageStore rocksDBMessageStore, StringBuilder diffResult, boolean checkAll) {
for (Map.Entry<Integer, ConsumeQueueInterface> queueEntry : queueMap.entrySet()) {
Integer queueId = queueEntry.getKey();
ConsumeQueueInterface jsonCq = queueEntry.getValue();
ConsumeQueueInterface kvCq = rocksDBMessageStore.getConsumeQueue(topic, queueId);
if (!checkAll) {
String format = String.format("\n[topic: %s, queue: %s] \n kvEarliest : %s | kvLatest : %s \n fileEarliest: %s | fileEarliest: %s ",
topic, queueId, kvCq.getEarliestUnit(), kvCq.getLatestUnit(), jsonCq.getEarliestUnit(), jsonCq.getLatestUnit());
diffResult.append(format).append("\n");
}
long maxFileOffsetInQueue = jsonCq.getMaxOffsetInQueue();
long minOffsetInQueue = kvCq.getMinOffsetInQueue();
for (long i = minOffsetInQueue; i < maxFileOffsetInQueue; i++) {
Pair<CqUnit, Long> fileCqUnit = jsonCq.getCqUnitAndStoreTime(i);
Pair<CqUnit, Long> kvCqUnit = kvCq.getCqUnitAndStoreTime(i);
if (fileCqUnit == null || kvCqUnit == null) {
diffResult.append(String.format("[topic: %s, queue: %s, offset: %s] \n kv : %s \n file: %s \n",
topic, queueId, i, kvCqUnit != null ? kvCqUnit.getObject1() : "null", fileCqUnit != null ? fileCqUnit.getObject1() : "null"));
return;
}
if (!checkCqUnitEqual(kvCqUnit.getObject1(), fileCqUnit.getObject1())) {
String diffInfo = String.format("[topic:%s, queue: %s offset: %s] \n file: %s \n kv: %s",
topic, queueId, i, kvCqUnit.getObject1(), fileCqUnit.getObject1());
LOGGER.error(diffInfo);
diffResult.append(diffInfo).append("\n");
return;
}
}
}
}
@Override
public boolean rejectRequest() {
return false;
@@ -3305,4 +3377,20 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
}
return false;
}
private boolean checkCqUnitEqual(CqUnit cqUnit1, CqUnit cqUnit2) {
if (cqUnit1.getQueueOffset() != cqUnit2.getQueueOffset()) {
return false;
}
if (cqUnit1.getSize() != cqUnit2.getSize()) {
return false;
}
if (cqUnit1.getPos() != cqUnit2.getPos()) {
return false;
}
if (cqUnit1.getBatchNum() != cqUnit2.getBatchNum()) {
return false;
}
return cqUnit1.getTagsCode() == cqUnit2.getTagsCode();
}
}
broker/src/main/java/org/apache/rocketmq/broker/subscription/RocksDBSubscriptionGroupManager.java
+20 -16
View File
@@ -19,6 +19,12 @@ package org.apache.rocketmq.broker.subscription;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import java.io.File;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.BiConsumer;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.RocksDBConfigManager;
import org.apache.rocketmq.common.UtilAll;
@@ -27,13 +33,6 @@ import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.rocksdb.RocksIterator;
import java.io.File;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.BiConsumer;
public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
protected RocksDBConfigManager rocksDBConfigManager;
@@ -79,28 +78,30 @@ public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
private boolean merge() {
if (!brokerController.getMessageStoreConfig().isTransferMetadataJsonToRocksdb()) {
log.info("The switch is off, no merge operation is needed.");
log.info("the switch transferMetadataJsonToRocksdb is off, no merge subGroup operation is needed.");
return true;
}
if (!UtilAll.isPathExists(this.configFilePath()) && !UtilAll.isPathExists(this.configFilePath() + ".bak")) {
log.info("json file and json back file not exist, so skip merge");
log.info("subGroup json file does not exist, so skip merge");
return true;
}
if (!super.load()) {
log.error("load group and forbidden info from json file error, startup will exit");
if (!super.loadDataVersion()) {
log.error("load json subGroup dataVersion error, startup will exit");
return false;
}
final ConcurrentMap<String, SubscriptionGroupConfig> groupTable = this.getSubscriptionGroupTable();
final ConcurrentMap<String, ConcurrentMap<String, Integer>> forbiddenTable = this.getForbiddenTable();
final DataVersion dataVersion = super.getDataVersion();
final DataVersion kvDataVersion = this.getDataVersion();
if (dataVersion.getCounter().get() > kvDataVersion.getCounter().get()) {
if (!super.load()) {
log.error("load group and forbidden info from json file error, startup will exit");
return false;
}
final ConcurrentMap<String, SubscriptionGroupConfig> groupTable = this.getSubscriptionGroupTable();
for (Map.Entry<String, SubscriptionGroupConfig> entry : groupTable.entrySet()) {
putSubscriptionGroupConfig(entry.getValue());
log.info("import subscription config to rocksdb, group={}", entry.getValue());
}
final ConcurrentMap<String, ConcurrentMap<String, Integer>> forbiddenTable = this.getForbiddenTable();
for (Map.Entry<String, ConcurrentMap<String, Integer>> entry : forbiddenTable.entrySet()) {
try {
this.rocksDBConfigManager.updateForbidden(entry.getKey(), JSON.toJSONString(entry.getValue()));
@@ -110,8 +111,10 @@ public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
return false;
}
}
this.rocksDBConfigManager.getKvDataVersion().assignNewOne(dataVersion);
this.getDataVersion().assignNewOne(dataVersion);
updateDataVersion();
} else {
log.info("dataVersion is not greater than kvDataVersion, no need to merge group metaData, dataVersion={}, kvDataVersion={}", dataVersion, kvDataVersion);
}
log.info("finish marge subscription config from json file and merge to rocksdb");
this.persist();
@@ -196,6 +199,7 @@ public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
try {
rocksDBConfigManager.updateKvDataVersion();
} catch (Exception e) {
log.error("update group config dataVersion error", e);
throw new RuntimeException(e);
}
}
broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java
+20
View File
@@ -334,6 +334,26 @@ public class SubscriptionGroupManager extends ConfigManager {
return dataVersion;
}
public boolean loadDataVersion() {
String fileName = null;
try {
fileName = this.configFilePath();
String jsonString = MixAll.file2String(fileName);
if (jsonString != null) {
SubscriptionGroupManager obj = RemotingSerializable.fromJson(jsonString, SubscriptionGroupManager.class);
if (obj != null) {
this.dataVersion.assignNewOne(obj.dataVersion);
this.printLoadDataWhenFirstBoot(obj);
log.info("load subGroup dataVersion success,{},{}", fileName, obj.dataVersion);
}
}
return true;
} catch (Exception e) {
log.error("load subGroup dataVersion failed" + fileName, e);
return false;
}
}
public void deleteSubscriptionGroupConfig(final String groupName) {
SubscriptionGroupConfig old = removeSubscriptionGroupConfig(groupName);
this.forbiddenTable.remove(groupName);
broker/src/main/java/org/apache/rocketmq/broker/topic/RocksDBTopicConfigManager.java
+16 -10
View File
@@ -18,6 +18,9 @@ package org.apache.rocketmq.broker.topic;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import java.io.File;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.RocksDBConfigManager;
import org.apache.rocketmq.common.TopicConfig;
@@ -25,10 +28,6 @@ import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.utils.DataConverter;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import java.io.File;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
public class RocksDBTopicConfigManager extends TopicConfigManager {
protected RocksDBConfigManager rocksDBConfigManager;
@@ -60,29 +59,35 @@ public class RocksDBTopicConfigManager extends TopicConfigManager {
private boolean merge() {
if (!brokerController.getMessageStoreConfig().isTransferMetadataJsonToRocksdb()) {
log.info("The switch is off, no merge operation is needed.");
log.info("the switch transferMetadataJsonToRocksdb is off, no merge topic operation is needed.");
return true;
}
if (!UtilAll.isPathExists(this.configFilePath()) && !UtilAll.isPathExists(this.configFilePath() + ".bak")) {
log.info("json file and json back file not exist, so skip merge");
log.info("topic json file does not exist, so skip merge");
return true;
}
if (!super.load()) {
log.error("load topic config from json file error, startup will exit");
if (!super.loadDataVersion()) {
log.error("load json topic dataVersion error, startup will exit");
return false;
}
final ConcurrentMap<String, TopicConfig> topicConfigTable = this.getTopicConfigTable();
final DataVersion dataVersion = super.getDataVersion();
final DataVersion kvDataVersion = this.getDataVersion();
if (dataVersion.getCounter().get() > kvDataVersion.getCounter().get()) {
if (!super.load()) {
log.error("load topic config from json file error, startup will exit");
return false;
}
final ConcurrentMap<String, TopicConfig> topicConfigTable = this.getTopicConfigTable();
for (Map.Entry<String, TopicConfig> entry : topicConfigTable.entrySet()) {
putTopicConfig(entry.getValue());
log.info("import topic config to rocksdb, topic={}", entry.getValue());
}
this.rocksDBConfigManager.getKvDataVersion().assignNewOne(dataVersion);
this.getDataVersion().assignNewOne(dataVersion);
updateDataVersion();
} else {
log.info("dataVersion is not greater than kvDataVersion, no need to merge topic metaData, dataVersion={}, kvDataVersion={}", dataVersion, kvDataVersion);
}
log.info("finish read topic config from json file and merge to rocksdb");
this.persist();
@@ -150,6 +155,7 @@ public class RocksDBTopicConfigManager extends TopicConfigManager {
try {
rocksDBConfigManager.updateKvDataVersion();
} catch (Exception e) {
log.error("update topic config dataVersion error", e);
throw new RuntimeException(e);
}
}
broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
+20
View File
@@ -637,6 +637,26 @@ public class TopicConfigManager extends ConfigManager {
return encode(false);
}
public boolean loadDataVersion() {
String fileName = null;
try {
fileName = this.configFilePath();
String jsonString = MixAll.file2String(fileName);
if (jsonString != null) {
TopicConfigSerializeWrapper topicConfigSerializeWrapper =
TopicConfigSerializeWrapper.fromJson(jsonString, TopicConfigSerializeWrapper.class);
if (topicConfigSerializeWrapper != null) {
this.dataVersion.assignNewOne(topicConfigSerializeWrapper.getDataVersion());
log.info("load topic metadata dataVersion success {}, {}", fileName, topicConfigSerializeWrapper.getDataVersion());
}
}
return true;
} catch (Exception e) {
log.error("load topic metadata dataVersion failed" + fileName, e);
return false;
}
}
@Override
public String configFilePath() {
return BrokerPathConfigHelper.getTopicConfigPath(this.brokerController.getMessageStoreConfig().getStorePathRootDir());
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksdbTransferOffsetAndCqTest.java
+154
View File
@@ -0,0 +1,154 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.broker.offset;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.RocksDBMessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.rocksdb.RocksDBException;
@RunWith(MockitoJUnitRunner.class)
public class RocksdbTransferOffsetAndCqTest {
private final String basePath = Paths.get(System.getProperty("user.home"),
"unit-test-store", UUID.randomUUID().toString().substring(0, 16).toUpperCase()).toString();
private final String topic = "topic";
private final String group = "group";
private final String clientHost = "clientHost";
private final int queueId = 1;
private RocksDBConsumerOffsetManager rocksdbConsumerOffsetManager;
private ConsumerOffsetManager consumerOffsetManager;
private DefaultMessageStore defaultMessageStore;
@Mock
private BrokerController brokerController;
@Before
public void init() throws IOException {
if (notToBeExecuted()) {
return;
}
BrokerConfig brokerConfig = new BrokerConfig();
brokerConfig.setConsumerOffsetUpdateVersionStep(10);
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(basePath);
messageStoreConfig.setTransferOffsetJsonToRocksdb(true);
messageStoreConfig.setRocksdbCQDoubleWriteEnable(true);
Mockito.lenient().when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
Mockito.lenient().when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
defaultMessageStore = new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("aaa", true), null,
brokerConfig, new ConcurrentHashMap<String, TopicConfig>());
defaultMessageStore.enableRocksdbCQWrite();
defaultMessageStore.loadCheckPoint();
consumerOffsetManager = new ConsumerOffsetManager(brokerController);
consumerOffsetManager.load();
rocksdbConsumerOffsetManager = new RocksDBConsumerOffsetManager(brokerController);
}
@Test
public void testTransferOffset() {
if (notToBeExecuted()) {
return;
}
for (int i = 0; i < 200; i++) {
consumerOffsetManager.commitOffset(clientHost, group, topic, queueId, i);
}
ConcurrentMap<String, ConcurrentMap<Integer, Long>> offsetTable = consumerOffsetManager.getOffsetTable();
ConcurrentMap<Integer, Long> map = offsetTable.get(topic + "@" + group);
Assert.assertTrue(MapUtils.isNotEmpty(map));
Long offset = map.get(queueId);
Assert.assertEquals(199L, (long) offset);
long offsetDataVersion = consumerOffsetManager.getDataVersion().getCounter().get();
Assert.assertEquals(20L, offsetDataVersion);
consumerOffsetManager.persist();
boolean loadResult = rocksdbConsumerOffsetManager.load();
Assert.assertTrue(loadResult);
ConcurrentMap<String, ConcurrentMap<Integer, Long>> rocksdbOffsetTable = rocksdbConsumerOffsetManager.getOffsetTable();
ConcurrentMap<Integer, Long> rocksdbMap = rocksdbOffsetTable.get(topic + "@" + group);
Assert.assertTrue(MapUtils.isNotEmpty(rocksdbMap));
Long aLong1 = rocksdbMap.get(queueId);
Assert.assertEquals(199L, (long) aLong1);
long rocksdbOffset = rocksdbConsumerOffsetManager.getDataVersion().getCounter().get();
Assert.assertEquals(21L, rocksdbOffset);
}
@Test
public void testRocksdbCqWrite() throws RocksDBException {
if (notToBeExecuted()) {
return;
}
RocksDBMessageStore kvStore = defaultMessageStore.getRocksDBMessageStore();
ConsumeQueueStoreInterface store = kvStore.getConsumeQueueStore();
ConsumeQueueInterface rocksdbCq = defaultMessageStore.getRocksDBMessageStore().findConsumeQueue(topic, queueId);
ConsumeQueueInterface fileCq = defaultMessageStore.findConsumeQueue(topic, queueId);
for (int i = 0; i < 200; i++) {
DispatchRequest request = new DispatchRequest(topic, queueId, i, 200, 0, System.currentTimeMillis(), i, "", "", 0, 0, new HashMap<>());
fileCq.putMessagePositionInfoWrapper(request);
store.putMessagePositionInfoWrapper(request);
}
Pair<CqUnit, Long> unit = rocksdbCq.getCqUnitAndStoreTime(100);
Pair<CqUnit, Long> unit1 = fileCq.getCqUnitAndStoreTime(100);
Assert.assertTrue(unit.getObject1().getPos() == unit1.getObject1().getPos());
}
private boolean notToBeExecuted() {
return MixAll.isMac();
}
}
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+15
View File
@@ -113,6 +113,7 @@ import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.CheckClientRequestBody;
import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
@@ -148,6 +149,7 @@ import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.AddBrokerRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.CheckRocksdbCqWriteProgressRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.CloneGroupOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumerSendMsgBackRequestHeader;
@@ -3017,6 +3019,19 @@ public class MQClientAPIImpl implements NameServerUpdateCallback, StartAndShutdo
throw new MQClientException(response.getCode(), response.getRemark());
}
public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(final String brokerAddr, final String topic, final long timeoutMillis) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
CheckRocksdbCqWriteProgressRequestHeader header = new CheckRocksdbCqWriteProgressRequestHeader();
header.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CHECK_ROCKSDB_CQ_WRITE_PROGRESS, header);
RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, timeoutMillis);
assert response != null;
if (ResponseCode.SUCCESS == response.getCode()) {
return CheckRocksdbCqWriteProgressResponseBody.decode(response.getBody(), CheckRocksdbCqWriteProgressResponseBody.class);
}
throw new MQClientException(response.getCode(), response.getRemark());
}
public void checkClientInBroker(final String brokerAddr, final String consumerGroup,
final String clientId, final SubscriptionData subscriptionData,
final long timeoutMillis)
common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java
+12 -11
View File
@@ -17,6 +17,15 @@
package org.apache.rocketmq.common.config;
import com.google.common.collect.Maps;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.DataConverter;
@@ -40,16 +49,6 @@ import org.rocksdb.Status;
import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static org.rocksdb.RocksDB.NOT_FOUND;
public abstract class AbstractRocksDBStorage {
@@ -495,7 +494,9 @@ public abstract class AbstractRocksDBStorage {
String blocksPinnedByIteratorMemUsage = this.db.getProperty("rocksdb.block-cache-pinned-usage");
logger.info("MemUsage. blockCache: {}, indexesAndFilterBlock: {}, memtable: {}, blocksPinnedByIterator: {}",
blockCacheMemUsage, indexesAndFilterBlockMemUsage, memTableMemUsage, blocksPinnedByIteratorMemUsage);
} catch (Exception ignored) {
} catch (Exception e) {
logger.error("statRocksdb Failed. {}", this.dbPath, e);
throw new RuntimeException(e);
}
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java
+1
View File
@@ -217,6 +217,7 @@ public class RequestCode {
public static final int GET_SUBSCRIPTIONGROUP_CONFIG = 352;
public static final int UPDATE_AND_GET_GROUP_FORBIDDEN = 353;
public static final int CHECK_ROCKSDB_CQ_WRITE_PROGRESS = 354;
public static final int LITE_PULL_MESSAGE = 361;
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/CheckRocksdbCqWriteProgressResponseBody.java
+35
View File
@@ -0,0 +1,35 @@
/*
* 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.remoting.protocol.body;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class CheckRocksdbCqWriteProgressResponseBody extends RemotingSerializable {
String diffResult;
public String getDiffResult() {
return diffResult;
}
public void setDiffResult(String diffResult) {
this.diffResult = diffResult;
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckRocksdbCqWriteProgressRequestHeader.java
+47
View File
@@ -0,0 +1,47 @@
/*
* 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.remoting.protocol.header;
import org.apache.rocketmq.common.action.Action;
import org.apache.rocketmq.common.action.RocketMQAction;
import org.apache.rocketmq.common.resource.ResourceType;
import org.apache.rocketmq.common.resource.RocketMQResource;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RequestCode;
@RocketMQAction(value = RequestCode.CHECK_ROCKSDB_CQ_WRITE_PROGRESS, action = Action.GET)
public class CheckRocksdbCqWriteProgressRequestHeader implements CommandCustomHeader {
@CFNotNull
@RocketMQResource(ResourceType.TOPIC)
private String topic;
@Override
public void checkFields() throws RemotingCommandException {
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
}
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+35 -7
View File
@@ -163,11 +163,13 @@ public class DefaultMessageStore implements MessageStore {
private volatile boolean shutdown = true;
protected boolean notifyMessageArriveInBatch = false;
private StoreCheckpoint storeCheckpoint;
protected StoreCheckpoint storeCheckpoint;
private TimerMessageStore timerMessageStore;
private final LinkedList<CommitLogDispatcher> dispatcherList;
private RocksDBMessageStore rocksDBMessageStore;
private RandomAccessFile lockFile;
private FileLock lock;
@@ -354,12 +356,7 @@ public class DefaultMessageStore implements MessageStore {
}
if (result) {
this.storeCheckpoint =
new StoreCheckpoint(
StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
this.masterFlushedOffset = this.storeCheckpoint.getMasterFlushedOffset();
setConfirmOffset(this.storeCheckpoint.getConfirmPhyOffset());
loadCheckPoint();
result = this.indexService.load(lastExitOK);
this.recover(lastExitOK);
LOGGER.info("message store recover end, and the max phy offset = {}", this.getMaxPhyOffset());
@@ -381,6 +378,14 @@ public class DefaultMessageStore implements MessageStore {
return result;
}
public void loadCheckPoint() throws IOException {
this.storeCheckpoint =
new StoreCheckpoint(
StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
this.masterFlushedOffset = this.storeCheckpoint.getMasterFlushedOffset();
setConfirmOffset(this.storeCheckpoint.getConfirmPhyOffset());
}
/**
* @throws Exception
*/
@@ -511,6 +516,10 @@ public class DefaultMessageStore implements MessageStore {
this.compactionService.shutdown();
}
if (messageStoreConfig.isRocksdbCQDoubleWriteEnable()) {
this.rocksDBMessageStore.consumeQueueStore.shutdown();
}
this.flushConsumeQueueService.shutdown();
this.allocateMappedFileService.shutdown();
this.storeCheckpoint.flush();
@@ -3251,6 +3260,17 @@ public class DefaultMessageStore implements MessageStore {
}
}
public void enableRocksdbCQWrite() {
try {
RocksDBMessageStore store = new RocksDBMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, this.topicConfigTable);
this.rocksDBMessageStore = store;
store.loadAndStartConsumerServiceOnly();
addDispatcher(store.getDispatcherBuildRocksdbConsumeQueue());
} catch (Exception e) {
LOGGER.error("enableRocksdbCqWrite error", e);
}
}
public int getMaxDelayLevel() {
return maxDelayLevel;
}
@@ -3338,4 +3358,12 @@ public class DefaultMessageStore implements MessageStore {
public long getReputFromOffset() {
return this.reputMessageService.getReputFromOffset();
}
public RocksDBMessageStore getRocksDBMessageStore() {
return this.rocksDBMessageStore;
}
public ConsumeQueueStoreInterface getConsumeQueueStore() {
return consumeQueueStore;
}
}
store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java
+41 -3
View File
@@ -16,16 +16,16 @@
*/
package org.apache.rocketmq.store;
import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.metrics.Meter;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Supplier;
import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.metrics.Meter;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager;
@@ -39,6 +39,8 @@ import org.rocksdb.RocksDBException;
public class RocksDBMessageStore extends DefaultMessageStore {
private CommitLogDispatcherBuildRocksdbConsumeQueue dispatcherBuildRocksdbConsumeQueue;
public RocksDBMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig, final ConcurrentMap<String, TopicConfig> topicConfigTable) throws
IOException {
@@ -178,4 +180,40 @@ public class RocksDBMessageStore extends DefaultMessageStore {
// Also add some metrics for rocksdb's monitoring.
RocksDBStoreMetricsManager.init(meter, attributesBuilderSupplier, this);
}
public CommitLogDispatcherBuildRocksdbConsumeQueue getDispatcherBuildRocksdbConsumeQueue() {
return dispatcherBuildRocksdbConsumeQueue;
}
class CommitLogDispatcherBuildRocksdbConsumeQueue implements CommitLogDispatcher {
@Override
public void dispatch(DispatchRequest request) throws RocksDBException {
final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());
switch (tranType) {
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
putMessagePositionInfo(request);
break;
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
break;
}
}
}
public void loadAndStartConsumerServiceOnly() {
try {
this.dispatcherBuildRocksdbConsumeQueue = new CommitLogDispatcherBuildRocksdbConsumeQueue();
boolean loadResult = this.consumeQueueStore.load();
if (!loadResult) {
throw new RuntimeException("load consume queue failed");
}
super.loadCheckPoint();
this.consumeQueueStore.start();
} catch (Exception e) {
ERROR_LOG.error("loadAndStartConsumerServiceOnly error", e);
throw new RuntimeException(e);
}
}
}
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+31
View File
@@ -424,6 +424,37 @@ public class MessageStoreConfig {
private boolean putConsumeQueueDataByFileChannel = true;
private boolean transferOffsetJsonToRocksdb = false;
private boolean rocksdbCQDoubleWriteEnable = false;
private boolean enableBatchWriteKvCq = true;
public boolean isEnableBatchWriteKvCq() {
return enableBatchWriteKvCq;
}
public void setEnableBatchWriteKvCq(boolean enableBatchWriteKvCq) {
this.enableBatchWriteKvCq = enableBatchWriteKvCq;
}
public boolean isRocksdbCQDoubleWriteEnable() {
return rocksdbCQDoubleWriteEnable;
}
public void setRocksdbCQDoubleWriteEnable(boolean rocksdbWriteEnable) {
this.rocksdbCQDoubleWriteEnable = rocksdbWriteEnable;
}
public boolean isTransferOffsetJsonToRocksdb() {
return transferOffsetJsonToRocksdb;
}
public void setTransferOffsetJsonToRocksdb(boolean transferOffsetJsonToRocksdb) {
this.transferOffsetJsonToRocksdb = transferOffsetJsonToRocksdb;
}
public boolean isEnabledAppendPropCRC() {
return enabledAppendPropCRC;
}
store/src/main/java/org/apache/rocketmq/store/queue/CqUnit.java
+1
View File
@@ -109,6 +109,7 @@ public class CqUnit {
", size=" + size +
", pos=" + pos +
", batchNum=" + batchNum +
", tagsCode=" + tagsCode +
", compactedOffset=" + compactedOffset +
'}';
}
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java
+1 -2
View File
@@ -18,7 +18,6 @@ package org.apache.rocketmq.store.queue;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.attribute.CQType;
@@ -311,7 +310,7 @@ public class RocksDBConsumeQueue implements ConsumeQueueInterface {
public CqUnit getLatestUnit() {
try {
long maxOffset = this.messageStore.getQueueStore().getMaxOffsetInQueue(topic, queueId);
return get(maxOffset);
return get(maxOffset > 0 ? maxOffset - 1 : maxOffset);
} catch (RocksDBException e) {
ERROR_LOG.error("getLatestUnit Failed. topic: {}, queueId: {}, {}", topic, queueId, e.getMessage());
}
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
+6 -4
View File
@@ -28,7 +28,6 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.BoundaryType;
@@ -78,6 +77,8 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
private final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap;
private volatile boolean isCQError = false;
private boolean enableBatchWriteKvCq;
public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
super(messageStore);
@@ -87,6 +88,7 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
this.rocksDBConsumeQueueOffsetTable = new RocksDBConsumeQueueOffsetTable(rocksDBConsumeQueueTable, rocksDBStorage, messageStore);
this.writeBatch = new WriteBatch();
this.enableBatchWriteKvCq = messageStoreConfig.isEnableBatchWriteKvCq();
this.bufferDRList = new ArrayList(BATCH_SIZE);
this.cqBBPairList = new ArrayList(BATCH_SIZE);
this.offsetBBPairList = new ArrayList(BATCH_SIZE);
@@ -164,12 +166,12 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
@Override
public void putMessagePositionInfoWrapper(DispatchRequest request) throws RocksDBException {
if (request == null || this.bufferDRList.size() >= BATCH_SIZE) {
putMessagePosition();
}
if (request != null) {
this.bufferDRList.add(request);
}
if (request == null || !enableBatchWriteKvCq || this.bufferDRList.size() >= BATCH_SIZE) {
putMessagePosition();
}
}
public void putMessagePosition() throws RocksDBException {
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+7
View File
@@ -52,6 +52,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -771,6 +772,12 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
);
}
@Override
public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.defaultMQAdminExtImpl.checkRocksdbCqWriteProgress(brokerAddr, topic);
}
@Override
public boolean resumeCheckHalfMessage(String topic,
String msgId)
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+7
View File
@@ -90,6 +90,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -1817,6 +1818,12 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return this.mqClientInstance.getMQClientAPIImpl().queryConsumeQueue(brokerAddr, topic, queueId, index, count, consumerGroup, timeoutMillis);
}
@Override
public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.mqClientInstance.getMQClientAPIImpl().checkRocksdbCqWriteProgress(brokerAddr, topic, timeoutMillis);
}
@Override
public boolean resumeCheckHalfMessage(final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+3
View File
@@ -48,6 +48,7 @@ import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -148,6 +149,8 @@ public interface MQAdminExt extends MQAdmin {
final String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException;
CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
ConsumeStats examineConsumeStats(final String consumerGroup,
final String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException;
tools/src/main/java/org/apache/rocketmq/tools/command/export/ExportMetadataInRocksDBCommand.java
+3 -1
View File
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.tools.command.export;
import com.alibaba.fastjson.JSONObject;
@@ -77,6 +78,7 @@ public class ExportMetadataInRocksDBCommand implements SubCommand {
}
String configType = commandLine.getOptionValue("configType").trim().toLowerCase();
path += "/" + configType;
boolean jsonEnable = false;
if (commandLine.hasOption("jsonEnable")) {
@@ -86,7 +88,7 @@ public class ExportMetadataInRocksDBCommand implements SubCommand {
ConfigRocksDBStorage kvStore = new ConfigRocksDBStorage(path, true /* readOnly */);
if (!kvStore.start()) {
System.out.print("RocksDB load error, path=" + path + "\n");
System.out.printf("RocksDB load error, path=%s\n" , path);
return;
}
tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java
+97
View File
@@ -0,0 +1,97 @@
/*
* 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.tools.command.queue;
import java.util.Map;
import java.util.Set;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.body.CheckRocksdbCqWriteProgressResponseBody;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommand;
public class CheckRocksdbCqWriteProgressCommand implements SubCommand {
@Override
public String commandName() {
return "checkRocksdbCqWriteProgressCommandCommand";
}
@Override
public String commandDesc() {
return "check if rocksdb cq is same as file cq";
}
@Override
public Options buildCommandlineOptions(Options options) {
Option opt = new Option("c", "cluster", true, "cluster name");
opt.setRequired(true);
options.addOption(opt);
opt = new Option("n", "nameserverAddr", true, "nameserverAddr");
opt.setRequired(true);
options.addOption(opt);
opt = new Option("t", "topic", true, "topic name");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@Override
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
defaultMQAdminExt.setNamesrvAddr(StringUtils.trim(commandLine.getOptionValue('n')));
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : "";
String topic = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : "";
try {
defaultMQAdminExt.start();
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
Map<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
Map<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
if (clusterAddrTable.get(clusterName) == null) {
System.out.print("clusterAddrTable is empty");
return;
}
for (Map.Entry<String, BrokerData> entry : brokerAddrTable.entrySet()) {
String brokerName = entry.getKey();
BrokerData brokerData = entry.getValue();
String brokerAddr = brokerData.getBrokerAddrs().get(0L);
CheckRocksdbCqWriteProgressResponseBody body = defaultMQAdminExt.checkRocksdbCqWriteProgress(brokerAddr, topic);
if (StringUtils.isNotBlank(topic)) {
System.out.printf(body.getDiffResult());
} else {
System.out.printf(brokerName + " | " + brokerAddr + " | " + body.getDiffResult());
}
}
} catch (Exception e) {
throw new RuntimeException(this.getClass().getSimpleName() + " command failed", e);
} finally {
defaultMQAdminExt.shutdown();
}
}
}