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

[ISSUE #8822] Double write cq, reduce unnecessary switches (#8823)

Browse Source 
* Reduce unnecessary switches
This commit is contained in:
LetLetMe
2024-10-29 19:09:35 +08:00
committed by GitHub
parent a96a14f734
commit 0b24768500
21 changed files with 246 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/RocksDBConfigManager.java
+7 -7
View File
@@ -17,40 +17,40 @@
package org.apache.rocketmq.broker;
import com.alibaba.fastjson.JSON;
import java.nio.charset.StandardCharsets;
import java.util.function.BiConsumer;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.config.ConfigRocksDBStorage;
import org.apache.rocketmq.common.constant.LoggerName;
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.CompressionType;
import org.rocksdb.FlushOptions;
import org.rocksdb.RocksIterator;
import org.rocksdb.Statistics;
import org.rocksdb.WriteBatch;
import java.nio.charset.StandardCharsets;
import java.util.function.BiConsumer;
public class RocksDBConfigManager {
protected static final Logger BROKER_LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
public volatile boolean isStop = false;
public ConfigRocksDBStorage configRocksDBStorage = null;
private FlushOptions flushOptions = null;
private volatile long lastFlushMemTableMicroSecond = 0;
private final String filePath;
private final long memTableFlushInterval;
private final CompressionType compressionType;
private DataVersion kvDataVersion = new DataVersion();
public RocksDBConfigManager(String filePath, long memTableFlushInterval) {
public RocksDBConfigManager(String filePath, long memTableFlushInterval, CompressionType compressionType) {
this.filePath = filePath;
this.memTableFlushInterval = memTableFlushInterval;
this.compressionType = compressionType;
}
public boolean init() {
this.isStop = false;
this.configRocksDBStorage = new ConfigRocksDBStorage(filePath);
this.configRocksDBStorage = new ConfigRocksDBStorage(filePath, compressionType);
return this.configRocksDBStorage.start();
}
public boolean loadDataVersion() {
broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConsumerOffsetManager.java
+4 -5
View File
@@ -31,6 +31,7 @@ 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.CompressionType;
import org.rocksdb.WriteBatch;
public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
@@ -41,7 +42,9 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
public RocksDBConsumerOffsetManager(BrokerController brokerController) {
super(brokerController);
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs());
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs(),
CompressionType.getCompressionType(brokerController.getMessageStoreConfig().getRocksdbCompressionType()));
}
@Override
@@ -61,10 +64,6 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
}
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;
broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBSubscriptionGroupManager.java
+3 -5
View File
@@ -32,6 +32,7 @@ import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.utils.DataConverter;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.rocksdb.CompressionType;
import org.rocksdb.RocksIterator;
public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
@@ -40,7 +41,8 @@ public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
public RocksDBSubscriptionGroupManager(BrokerController brokerController) {
super(brokerController, false);
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs());
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs(),
CompressionType.getCompressionType(brokerController.getMessageStoreConfig().getRocksdbCompressionType()));
}
@Override
@@ -78,10 +80,6 @@ public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
private boolean merge() {
if (!brokerController.getMessageStoreConfig().isTransferMetadataJsonToRocksdb()) {
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("subGroup json file does not exist, so skip merge");
return true;
broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBTopicConfigManager.java
+3 -5
View File
@@ -28,6 +28,7 @@ import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.utils.DataConverter;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.rocksdb.CompressionType;
public class RocksDBTopicConfigManager extends TopicConfigManager {
@@ -35,7 +36,8 @@ public class RocksDBTopicConfigManager extends TopicConfigManager {
public RocksDBTopicConfigManager(BrokerController brokerController) {
super(brokerController, false);
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs());
this.rocksDBConfigManager = new RocksDBConfigManager(rocksdbConfigFilePath(), brokerController.getMessageStoreConfig().getMemTableFlushIntervalMs(),
CompressionType.getCompressionType(brokerController.getMessageStoreConfig().getRocksdbCompressionType()));
}
@Override
@@ -59,10 +61,6 @@ public class RocksDBTopicConfigManager extends TopicConfigManager {
}
private boolean merge() {
if (!brokerController.getMessageStoreConfig().isTransferMetadataJsonToRocksdb()) {
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("topic json file does not exist, so skip merge");
return true;
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+130 -68
View File
@@ -18,7 +18,6 @@ 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;
@@ -39,6 +38,9 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
@@ -65,7 +67,9 @@ import org.apache.rocketmq.broker.metrics.InvocationStatus;
import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.LockCallback;
import org.apache.rocketmq.common.MQVersion;
@@ -214,6 +218,7 @@ 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.StoreType;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.exception.ConsumeQueueException;
import org.apache.rocketmq.store.plugin.AbstractPluginMessageStore;
@@ -232,6 +237,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
protected final BrokerController brokerController;
protected Set<String> configBlackList = new HashSet<>();
private final ExecutorService asyncExecuteWorker = new ThreadPoolExecutor(0, 1, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
public AdminBrokerProcessor(final BrokerController brokerController) {
this.brokerController = brokerController;
@@ -467,76 +473,23 @@ 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);
private RemotingCommand checkRocksdbCqWriteProgress(ChannelHandlerContext ctx, RemotingCommand request) {
CheckRocksdbCqWriteResult result = new CheckRocksdbCqWriteResult();
result.setCheckStatus(CheckRocksdbCqWriteResult.CheckStatus.CHECK_IN_PROGRESS.getValue());
Runnable runnable = () -> {
try {
CheckRocksdbCqWriteResult checkResult = doCheckRocksdbCqWriteProgress(ctx, request);
LOGGER.info("checkRocksdbCqWriteProgress result: {}", JSON.toJSONString(checkResult));
} catch (Exception e) {
LOGGER.error("checkRocksdbCqWriteProgress error", e);
}
};
asyncExecuteWorker.submit(runnable);
RemotingCommand response = RemotingCommand.createResponseCommand(null);
response.setCode(ResponseCode.SUCCESS);
MessageStore messageStore = brokerController.getMessageStore();
DefaultMessageStore defaultMessageStore;
if (messageStore instanceof AbstractPluginMessageStore) {
defaultMessageStore = (DefaultMessageStore) ((AbstractPluginMessageStore) messageStore).getNext();
} else {
defaultMessageStore = (DefaultMessageStore) messageStore;
}
RocksDBMessageStore rocksDBMessageStore = defaultMessageStore.getRocksDBMessageStore();
if (!defaultMessageStore.getMessageStoreConfig().isRocksdbCQDoubleWriteEnable()) {
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult", "rocksdbCQWriteEnable is false, checkRocksdbCqWriteProgressCommand is invalid")));
return response;
}
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> cqTable = defaultMessageStore.getConsumeQueueTable();
StringBuilder diffResult = new StringBuilder();
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())));
}
response.setBody(JSON.toJSONBytes(result));
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 \n",
topic, queueId, i, kvCqUnit.getObject1(), fileCqUnit.getObject1());
LOGGER.error(diffInfo);
diffResult.append(diffInfo).append(System.lineSeparator());
return;
}
}
}
}
@Override
public boolean rejectRequest() {
return false;
@@ -3418,6 +3371,115 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
return false;
}
private CheckRocksdbCqWriteResult doCheckRocksdbCqWriteProgress(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
CheckRocksdbCqWriteProgressRequestHeader requestHeader = request.decodeCommandCustomHeader(CheckRocksdbCqWriteProgressRequestHeader.class);
String requestTopic = requestHeader.getTopic();
MessageStore messageStore = brokerController.getMessageStore();
DefaultMessageStore defaultMessageStore;
if (messageStore instanceof AbstractPluginMessageStore) {
defaultMessageStore = (DefaultMessageStore) ((AbstractPluginMessageStore) messageStore).getNext();
} else {
defaultMessageStore = (DefaultMessageStore) messageStore;
}
RocksDBMessageStore rocksDBMessageStore = defaultMessageStore.getRocksDBMessageStore();
CheckRocksdbCqWriteResult result = new CheckRocksdbCqWriteResult();
if (defaultMessageStore.getMessageStoreConfig().getStoreType().equals(StoreType.DEFAULT_ROCKSDB.getStoreType())) {
result.setCheckResult("storeType is DEFAULT_ROCKSDB, no need check");
result.setCheckStatus(CheckRocksdbCqWriteResult.CheckStatus.CHECK_OK.getValue());
return result;
}
if (!defaultMessageStore.getMessageStoreConfig().isRocksdbCQDoubleWriteEnable()) {
result.setCheckResult("rocksdbCQWriteEnable is false, checkRocksdbCqWriteProgressCommand is invalid");
result.setCheckStatus(CheckRocksdbCqWriteResult.CheckStatus.CHECK_NOT_OK.getValue());
return result;
}
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> cqTable = defaultMessageStore.getConsumeQueueTable();
StringBuilder diffResult = new StringBuilder();
try {
if (StringUtils.isNotBlank(requestTopic)) {
boolean checkResult = processConsumeQueuesForTopic(cqTable.get(requestTopic), requestTopic, rocksDBMessageStore, diffResult, true, requestHeader.getCheckStoreTime());
result.setCheckResult(diffResult.toString());
result.setCheckStatus(checkResult ? CheckRocksdbCqWriteResult.CheckStatus.CHECK_OK.getValue() : CheckRocksdbCqWriteResult.CheckStatus.CHECK_NOT_OK.getValue());
return result;
}
int successNum = 0;
int checkSize = 0;
for (Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> topicEntry : cqTable.entrySet()) {
boolean checkResult = processConsumeQueuesForTopic(topicEntry.getValue(), topicEntry.getKey(), rocksDBMessageStore, diffResult, false, requestHeader.getCheckStoreTime());
successNum += checkResult ? 1 : 0;
checkSize++;
}
// check all topic finish, all topic is ready, checkSize: 100, currentQueueNum: 110 -> ready (The currentQueueNum means when we do checking, new topics are added.)
// check all topic finish, success/all : 89/100, currentQueueNum: 110 -> not ready
boolean checkReady = successNum == checkSize;
String checkResultString = checkReady ? String.format("all topic is ready, checkSize: %s, currentQueueNum: %s", checkSize, cqTable.size()) :
String.format("success/all : %s/%s, currentQueueNum: %s", successNum, checkSize, cqTable.size());
diffResult.append("check all topic finish, ").append(checkResultString);
result.setCheckResult(diffResult.toString());
result.setCheckStatus(checkReady ? CheckRocksdbCqWriteResult.CheckStatus.CHECK_OK.getValue() : CheckRocksdbCqWriteResult.CheckStatus.CHECK_NOT_OK.getValue());
} catch (Exception e) {
LOGGER.error("CheckRocksdbCqWriteProgressCommand error", e);
result.setCheckResult(e.getMessage() + Arrays.toString(e.getStackTrace()));
result.setCheckStatus(CheckRocksdbCqWriteResult.CheckStatus.CHECK_ERROR.getValue());
}
return result;
}
private boolean processConsumeQueuesForTopic(ConcurrentMap<Integer, ConsumeQueueInterface> queueMap, String topic, RocksDBMessageStore rocksDBMessageStore, StringBuilder diffResult, boolean printDetail, long checkpointByStoreTime) {
boolean processResult = true;
for (Map.Entry<Integer, ConsumeQueueInterface> queueEntry : queueMap.entrySet()) {
Integer queueId = queueEntry.getKey();
ConsumeQueueInterface jsonCq = queueEntry.getValue();
ConsumeQueueInterface kvCq = rocksDBMessageStore.getConsumeQueue(topic, queueId);
if (printDetail) {
String format = String.format("[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 minOffsetByTime = 0L;
try {
minOffsetByTime = rocksDBMessageStore.getConsumeQueueStore().getOffsetInQueueByTime(topic, queueId, checkpointByStoreTime, BoundaryType.UPPER);
} catch (Exception e) {
// ignore
}
long minOffsetInQueue = kvCq.getMinOffsetInQueue();
long checkFrom = Math.max(minOffsetInQueue, minOffsetByTime);
long checkTo = jsonCq.getMaxOffsetInQueue() - 1;
/*
checkTo(maxOffsetInQueue - 1)
v
fileCq +------------------------------------------------------+
kvCq +----------------------------------------------+
^ ^
minOffsetInQueue minOffsetByTime
^
checkFrom = max(minOffsetInQueue, minOffsetByTime)
*/
// The latest message is earlier than the check time
Pair<CqUnit, Long> fileLatestCq = jsonCq.getCqUnitAndStoreTime(checkTo);
if (fileLatestCq != null) {
if (fileLatestCq.getObject2() < checkpointByStoreTime) {
continue;
}
}
for (long i = checkFrom; i <= checkTo; i++) {
Pair<CqUnit, Long> fileCqUnit = jsonCq.getCqUnitAndStoreTime(i);
Pair<CqUnit, Long> kvCqUnit = kvCq.getCqUnitAndStoreTime(i);
if (fileCqUnit == null || kvCqUnit == null || !checkCqUnitEqual(kvCqUnit.getObject1(), fileCqUnit.getObject1())) {
LOGGER.error(String.format("[topic: %s, queue: %s, offset: %s] \n file : %s \n kv : %s \n",
topic, queueId, i, kvCqUnit != null ? kvCqUnit.getObject1() : "null", fileCqUnit != null ? fileCqUnit.getObject1() : "null"));
processResult = false;
break;
}
}
}
return processResult;
}
private boolean checkCqUnitEqual(CqUnit cqUnit1, CqUnit cqUnit2) {
if (cqUnit1.getQueueOffset() != cqUnit2.getQueueOffset()) {
return false;
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksdbTransferOffsetAndCqTest.java
-1
View File
@@ -76,7 +76,6 @@ public class RocksdbTransferOffsetAndCqTest {
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);
broker/src/test/java/org/apache/rocketmq/broker/subscription/RocksdbGroupConfigTransferTest.java
-1
View File
@@ -68,7 +68,6 @@ public class RocksdbGroupConfigTransferTest {
Mockito.lenient().when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(basePath);
messageStoreConfig.setTransferMetadataJsonToRocksdb(true);
Mockito.lenient().when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
Mockito.lenient().when(brokerController.getMessageStore()).thenReturn(defaultMessageStore);
when(defaultMessageStore.getStateMachineVersion()).thenReturn(0L);
broker/src/test/java/org/apache/rocketmq/broker/topic/RocksdbTopicConfigManagerTest.java
-1
View File
@@ -72,7 +72,6 @@ public class RocksdbTopicConfigManagerTest {
when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(basePath);
messageStoreConfig.setTransferMetadataJsonToRocksdb(true);
when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
Mockito.lenient().when(brokerController.getMessageStore()).thenReturn(defaultMessageStore);
Mockito.lenient().when(defaultMessageStore.getStateMachineVersion()).thenReturn(0L);
broker/src/test/java/org/apache/rocketmq/broker/topic/RocksdbTopicConfigTransferTest.java
-1
View File
@@ -69,7 +69,6 @@ public class RocksdbTopicConfigTransferTest {
when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(basePath);
messageStoreConfig.setTransferMetadataJsonToRocksdb(true);
Mockito.lenient().when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
when(brokerController.getMessageStore()).thenReturn(defaultMessageStore);
when(defaultMessageStore.getStateMachineVersion()).thenReturn(0L);
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+4 -3
View File
@@ -56,6 +56,7 @@ import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.rpchook.NamespaceRpcHook;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
@@ -113,7 +114,6 @@ 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;
@@ -3019,15 +3019,16 @@ 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,
public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(final String brokerAddr, final String topic, final long checkStoreTime, final long timeoutMillis) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
CheckRocksdbCqWriteProgressRequestHeader header = new CheckRocksdbCqWriteProgressRequestHeader();
header.setTopic(topic);
header.setCheckStoreTime(checkStoreTime);
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);
return JSON.parseObject(response.getBody(), CheckRocksdbCqWriteResult.class);
}
throw new MQClientException(response.getCode(), response.getRemark());
}
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/CheckRocksdbCqWriteProgressResponseBody.java → common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java
+30 -8
View File
@@ -15,21 +15,43 @@
* limitations under the License.
*/
package org.apache.rocketmq.remoting.protocol.body;
package org.apache.rocketmq.common;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class CheckRocksdbCqWriteResult {
String checkResult;
public class CheckRocksdbCqWriteProgressResponseBody extends RemotingSerializable {
int checkStatus;
String diffResult;
public enum CheckStatus {
CHECK_OK(0),
CHECK_NOT_OK(1),
CHECK_IN_PROGRESS(2),
CHECK_ERROR(3);
public String getDiffResult() {
return diffResult;
private int value;
CheckStatus(int value) {
this.value = value;
}
public int getValue() {
return value;
}
}
public void setDiffResult(String diffResult) {
this.diffResult = diffResult;
public String getCheckResult() {
return checkResult;
}
public void setCheckResult(String checkResult) {
this.checkResult = checkResult;
}
public int getCheckStatus() {
return checkStatus;
}
public void setCheckStatus(int checkStatus) {
this.checkStatus = checkStatus;
}
}
common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java
+2 -1
View File
@@ -86,6 +86,7 @@ public abstract class AbstractRocksDBStorage {
protected final List<ColumnFamilyHandle> cfHandles = new ArrayList<>();
protected volatile boolean loaded;
protected CompressionType compressionType = CompressionType.LZ4_COMPRESSION;
private volatile boolean closed;
private final Semaphore reloadPermit = new Semaphore(1);
@@ -156,7 +157,7 @@ public abstract class AbstractRocksDBStorage {
protected void initCompactionOptions() {
this.compactionOptions = new CompactionOptions();
this.compactionOptions.setCompression(CompressionType.LZ4_COMPRESSION);
this.compactionOptions.setCompression(compressionType);
this.compactionOptions.setMaxSubcompactions(4);
this.compactionOptions.setOutputFileSizeLimit(4 * 1024 * 1024 * 1024L);
}
common/src/main/java/org/apache/rocketmq/common/config/ConfigRocksDBStorage.java
+9 -3
View File
@@ -25,6 +25,7 @@ import org.apache.rocketmq.common.UtilAll;
import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompressionType;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
@@ -37,12 +38,17 @@ public class ConfigRocksDBStorage extends AbstractRocksDBStorage {
protected ColumnFamilyHandle kvDataVersionFamilyHandle;
protected ColumnFamilyHandle forbiddenFamilyHandle;
public static final byte[] KV_DATA_VERSION_KEY = "kvDataVersionKey".getBytes(StandardCharsets.UTF_8);
public ConfigRocksDBStorage(final String dbPath) {
super(dbPath);
this.readOnly = false;
this(dbPath, false);
}
public ConfigRocksDBStorage(final String dbPath, CompressionType compressionType) {
this(dbPath, false);
this.compressionType = compressionType;
}
public ConfigRocksDBStorage(final String dbPath, boolean readOnly) {
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckRocksdbCqWriteProgressRequestHeader.java
+11
View File
@@ -32,6 +32,8 @@ public class CheckRocksdbCqWriteProgressRequestHeader implements CommandCustomHe
@RocketMQResource(ResourceType.TOPIC)
private String topic;
private long checkStoreTime;
@Override
public void checkFields() throws RemotingCommandException {
@@ -44,4 +46,13 @@ public class CheckRocksdbCqWriteProgressRequestHeader implements CommandCustomHe
public void setTopic(String topic) {
this.topic = topic;
}
public long getCheckStoreTime() {
return checkStoreTime;
}
public void setCheckStoreTime(long checkStoreTime) {
this.checkStoreTime = checkStoreTime;
}
}
store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java
+4
View File
@@ -173,6 +173,10 @@ public class RocksDBMessageStore extends DefaultMessageStore {
class CommitLogDispatcherBuildRocksdbConsumeQueue implements CommitLogDispatcher {
@Override
public void dispatch(DispatchRequest request) throws RocksDBException {
boolean enable = getMessageStoreConfig().isRocksdbCQDoubleWriteEnable();
if (!enable) {
return;
}
final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());
switch (tranType) {
case MessageSysFlag.TRANSACTION_NOT_TYPE:
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+12 -20
View File
@@ -22,6 +22,7 @@ import org.apache.rocketmq.common.annotation.ImportantField;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.StoreType;
import org.apache.rocketmq.store.queue.BatchConsumeQueue;
import org.rocksdb.CompressionType;
public class MessageStoreConfig {
@@ -106,8 +107,6 @@ public class MessageStoreConfig {
@ImportantField
private String storeType = StoreType.DEFAULT.getStoreType();
private boolean transferMetadataJsonToRocksdb = false;
// ConsumeQueue file size,default is 30W
private int mappedFileSizeConsumeQueue = 300000 * ConsumeQueue.CQ_STORE_UNIT_SIZE;
// enable consume queue ext
@@ -424,8 +423,6 @@ public class MessageStoreConfig {
private boolean putConsumeQueueDataByFileChannel = true;
private boolean transferOffsetJsonToRocksdb = false;
private boolean rocksdbCQDoubleWriteEnable = false;
/**
@@ -443,7 +440,17 @@ public class MessageStoreConfig {
*
* LZ4 is the recommended one.
*/
private String bottomMostCompressionTypeForConsumeQueueStore = "zstd";
private String bottomMostCompressionTypeForConsumeQueueStore = CompressionType.ZSTD_COMPRESSION.getLibraryName();
private String rocksdbCompressionType = CompressionType.LZ4_COMPRESSION.getLibraryName();
public String getRocksdbCompressionType() {
return rocksdbCompressionType;
}
public void setRocksdbCompressionType(String compressionType) {
this.rocksdbCompressionType = compressionType;
}
/**
* Spin number in the retreat strategy of spin lock
@@ -464,13 +471,6 @@ public class MessageStoreConfig {
this.rocksdbCQDoubleWriteEnable = rocksdbWriteEnable;
}
public boolean isTransferOffsetJsonToRocksdb() {
return transferOffsetJsonToRocksdb;
}
public void setTransferOffsetJsonToRocksdb(boolean transferOffsetJsonToRocksdb) {
this.transferOffsetJsonToRocksdb = transferOffsetJsonToRocksdb;
}
public boolean isEnabledAppendPropCRC() {
return enabledAppendPropCRC;
@@ -1894,14 +1894,6 @@ public class MessageStoreConfig {
this.putConsumeQueueDataByFileChannel = putConsumeQueueDataByFileChannel;
}
public boolean isTransferMetadataJsonToRocksdb() {
return transferMetadataJsonToRocksdb;
}
public void setTransferMetadataJsonToRocksdb(boolean transferMetadataJsonToRocksdb) {
this.transferMetadataJsonToRocksdb = transferMetadataJsonToRocksdb;
}
public String getBottomMostCompressionTypeForConsumeQueueStore() {
return bottomMostCompressionTypeForConsumeQueueStore;
}
store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java
+4 -1
View File
@@ -67,13 +67,16 @@ public class RocksDBOptionsFactory {
setCompressionSizePercent(-1);
String bottomMostCompressionTypeOpt = messageStore.getMessageStoreConfig()
.getBottomMostCompressionTypeForConsumeQueueStore();
String compressionTypeOpt = messageStore.getMessageStoreConfig()
.getRocksdbCompressionType();
CompressionType bottomMostCompressionType = CompressionType.getCompressionType(bottomMostCompressionTypeOpt);
CompressionType compressionType = CompressionType.getCompressionType(compressionTypeOpt);
return columnFamilyOptions.setMaxWriteBufferNumber(4).
setWriteBufferSize(128 * SizeUnit.MB).
setMinWriteBufferNumberToMerge(1).
setTableFormatConfig(blockBasedTableConfig).
setMemTableConfig(new SkipListMemTableConfig()).
setCompressionType(CompressionType.LZ4_COMPRESSION).
setCompressionType(compressionType).
setBottommostCompressionType(bottomMostCompressionType).
setNumLevels(7).
setCompactionStyle(CompactionStyle.UNIVERSAL).
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+3 -3
View File
@@ -26,6 +26,7 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
@@ -52,7 +53,6 @@ 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;
@@ -773,9 +773,9 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic)
public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, long checkStoreTime)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.defaultMQAdminExtImpl.checkRocksdbCqWriteProgress(brokerAddr, topic);
return this.defaultMQAdminExtImpl.checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime);
}
@Override
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+3 -3
View File
@@ -46,6 +46,7 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQClientManager;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
@@ -90,7 +91,6 @@ 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;
@@ -1819,9 +1819,9 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
public CheckRocksdbCqWriteProgressResponseBody checkRocksdbCqWriteProgress(String brokerAddr, String topic)
public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, long checkStoreTime)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.mqClientInstance.getMQClientAPIImpl().checkRocksdbCqWriteProgress(brokerAddr, topic, timeoutMillis);
return this.mqClientInstance.getMQClientAPIImpl().checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime, timeoutMillis);
}
@Override
tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+3 -2
View File
@@ -24,6 +24,7 @@ import java.util.Set;
import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
@@ -48,7 +49,6 @@ 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;
@@ -149,7 +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;
CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr, String topic, long checkStoreTime)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
ConsumeStats examineConsumeStats(final String consumerGroup,
final String topic) throws RemotingException, MQClientException,
tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java
+14 -7
View File
@@ -19,12 +19,13 @@ package org.apache.rocketmq.tools.command.queue;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
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.common.CheckRocksdbCqWriteResult;
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;
@@ -53,8 +54,11 @@ public class CheckRocksdbCqWriteProgressCommand implements SubCommand {
options.addOption(opt);
opt = new Option("t", "topic", true, "topic name");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("cf", "checkFrom", true, "check from time");
options.addOption(opt);
return options;
}
@@ -66,6 +70,10 @@ public class CheckRocksdbCqWriteProgressCommand implements SubCommand {
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() : "";
// The default check is 30 days
long checkStoreTime = commandLine.hasOption("cf")
? Long.parseLong(commandLine.getOptionValue("cf").trim())
: System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30L);
try {
defaultMQAdminExt.start();
@@ -80,14 +88,13 @@ public class CheckRocksdbCqWriteProgressCommand implements SubCommand {
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.print(body.getDiffResult());
CheckRocksdbCqWriteResult result = defaultMQAdminExt.checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime);
if (result.getCheckStatus() == CheckRocksdbCqWriteResult.CheckStatus.CHECK_ERROR.getValue()) {
System.out.print(brokerName + " check error, please check log... errInfo: " + result.getCheckResult());
} else {
System.out.print(brokerName + " | " + brokerAddr + " | \n" + body.getDiffResult());
System.out.print(brokerName + " check doing, please wait and get the result from log... \n");
}
}
} catch (Exception e) {
throw new RuntimeException(this.getClass().getSimpleName() + " command failed", e);
} finally {