代码拉取完成，页面将自动刷新
From ead3d905016d9db4785a46beaa555c7fafd4f9bb Mon Sep 17 00:00:00 2001
From: Dongyuan Pan <dongyuanpan0@gmail.com>
Date: Wed, 8 Nov 2023 10:40:52 +0800
Subject: [PATCH 1/2] [ISSUE #7511] Lock granularity issue causing LMQ message
loss (#7525)
* bug fix: assignOffset and increaseOffset in LMQ has concurrency issues in topicQueueLock, should be in putMessageLock
* fix MultiDispatchTest
* fix MultiDispatchTest
* fix unit test
---
.../common/message/MessageExtBrokerInner.java | 10 ++
.../org/apache/rocketmq/store/CommitLog.java | 94 ++++++++++++--
.../apache/rocketmq/store/ConsumeQueue.java | 44 +------
.../rocketmq/store/DefaultMessageStore.java | 1 -
.../rocketmq/store/MessageExtEncoder.java | 118 ++++++++++++++++--
.../apache/rocketmq/store/MultiDispatch.java | 77 ++++++++++++
.../queue/AbstractConsumeQueueStore.java | 10 ++
.../store/queue/ConsumeQueueInterface.java | 1 -
.../queue/ConsumeQueueStoreInterface.java | 14 +++
...iDispatch.java => MultiDispatchUtils.java} | 17 +--
.../store/queue/QueueOffsetOperator.java | 6 +-
.../store/queue/RocksDBConsumeQueue.java | 42 -------
.../rocketmq/store/AppendCallbackTest.java | 6 +-
.../rocketmq/store/AppendPropCRCTest.java | 5 +-
.../rocketmq/store/MultiDispatchTest.java | 12 +-
.../rocketmq/store/kv/CompactionLogTest.java | 2 +-
16 files changed, 322 insertions(+), 137 deletions(-)
create mode 100644 store/src/main/java/org/apache/rocketmq/store/MultiDispatch.java
rename store/src/main/java/org/apache/rocketmq/store/queue/{MultiDispatch.java => MultiDispatchUtils.java} (78%)
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageExtBrokerInner.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageExtBrokerInner.java
index 52501dbca..147f23f12 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/MessageExtBrokerInner.java
+++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageExtBrokerInner.java
@@ -28,6 +28,8 @@ public class MessageExtBrokerInner extends MessageExt {
private ByteBuffer encodedBuff;
+ private volatile boolean encodeCompleted;
+
private MessageVersion version = MessageVersion.MESSAGE_VERSION_V1;
public ByteBuffer getEncodedBuff() {
@@ -92,4 +94,12 @@ public class MessageExtBrokerInner extends MessageExt {
this.setPropertiesString(MessageDecoder.messageProperties2String(this.getProperties()));
}
}
+
+ public boolean isEncodeCompleted() {
+ return encodeCompleted;
+ }
+
+ public void setEncodeCompleted(boolean encodeCompleted) {
+ this.encodeCompleted = encodeCompleted;
+ }
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 6c3afde70..35c1d0e2d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -35,6 +35,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import com.sun.jna.NativeLong;
import com.sun.jna.Pointer;
+import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.SystemClock;
@@ -56,6 +57,7 @@ import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.MessageExtEncoder.PutMessageThreadLocal;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.FlushDiskType;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.HAService;
import org.apache.rocketmq.store.ha.autoswitch.AutoSwitchHAService;
import org.apache.rocketmq.store.logfile.MappedFile;
@@ -101,6 +103,7 @@ public class CommitLog implements Swappable {
protected int commitLogSize;
private final boolean enabledAppendPropCRC;
+ protected final MultiDispatch multiDispatch;
public CommitLog(final DefaultMessageStore messageStore) {
String storePath = messageStore.getMessageStoreConfig().getStorePathCommitLog();
@@ -119,13 +122,11 @@ public class CommitLog implements Swappable {
this.flushManager = new DefaultFlushManager();
this.coldDataCheckService = new ColdDataCheckService();
- this.appendMessageCallback = new DefaultAppendMessageCallback();
+ this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig());
putMessageThreadLocal = new ThreadLocal<PutMessageThreadLocal>() {
@Override
protected PutMessageThreadLocal initialValue() {
- return new PutMessageThreadLocal(
- defaultMessageStore.getMessageStoreConfig().getMaxMessageSize(),
- defaultMessageStore.getMessageStoreConfig().isEnabledAppendPropCRC());
+ return new PutMessageThreadLocal(defaultMessageStore.getMessageStoreConfig());
}
};
this.putMessageLock = messageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
@@ -137,6 +138,8 @@ public class CommitLog implements Swappable {
this.commitLogSize = messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
this.enabledAppendPropCRC = messageStore.getMessageStoreConfig().isEnabledAppendPropCRC();
+
+ this.multiDispatch = new MultiDispatch(defaultMessageStore);
}
public void setFullStorePaths(Set<String> fullStorePaths) {
@@ -1830,15 +1833,84 @@ public class CommitLog implements Swappable {
// Store the message content
private final ByteBuffer msgStoreItemMemory;
private final int crc32ReservedLength = CommitLog.CRC32_RESERVED_LEN;
+ private final MessageStoreConfig messageStoreConfig;
- DefaultAppendMessageCallback() {
+ DefaultAppendMessageCallback(MessageStoreConfig messageStoreConfig) {
this.msgStoreItemMemory = ByteBuffer.allocate(END_FILE_MIN_BLANK_LENGTH);
+ this.messageStoreConfig = messageStoreConfig;
+ }
+
+ public AppendMessageResult handlePropertiesForLmqMsg(ByteBuffer preEncodeBuffer, final MessageExtBrokerInner msgInner) {
+ if (msgInner.isEncodeCompleted()) {
+ return null;
+ }
+
+ multiDispatch.wrapMultiDispatch(msgInner);
+
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+
+ final byte[] propertiesData =
+ msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
+
+ boolean needAppendLastPropertySeparator = enabledAppendPropCRC && propertiesData != null && propertiesData.length > 0
+ && propertiesData[propertiesData.length - 1] != MessageDecoder.PROPERTY_SEPARATOR;
+
+ final int propertiesLength = (propertiesData == null ? 0 : propertiesData.length) + (needAppendLastPropertySeparator ? 1 : 0) + crc32ReservedLength;
+
+ if (propertiesLength > Short.MAX_VALUE) {
+ log.warn("putMessage message properties length too long. length={}", propertiesData.length);
+ return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
+ }
+
+ int msgLenWithoutProperties = preEncodeBuffer.getInt(0);
+
+ int msgLen = msgLenWithoutProperties + 2 + propertiesLength;
+
+ // Exceeds the maximum message
+ if (msgLen > this.messageStoreConfig.getMaxMessageSize()) {
+ log.warn("message size exceeded, msg total size: " + msgLen + ", maxMessageSize: " + this.messageStoreConfig.getMaxMessageSize());
+ return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
+ }
+
+ // Back filling total message length
+ preEncodeBuffer.putInt(0, msgLen);
+ // Modify position to msgLenWithoutProperties
+ preEncodeBuffer.position(msgLenWithoutProperties);
+
+ preEncodeBuffer.putShort((short) propertiesLength);
+
+ if (propertiesLength > crc32ReservedLength) {
+ preEncodeBuffer.put(propertiesData);
+ }
+
+ if (needAppendLastPropertySeparator) {
+ preEncodeBuffer.put((byte) MessageDecoder.PROPERTY_SEPARATOR);
+ }
+ // 18 CRC32
+ preEncodeBuffer.position(preEncodeBuffer.position() + crc32ReservedLength);
+
+ msgInner.setEncodeCompleted(true);
+
+ return null;
}
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {
// STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
+ ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
+ boolean isMultiDispatchMsg = messageStoreConfig.isEnableMultiDispatch() && CommitLog.isMultiDispatchMsg(msgInner);
+ if (isMultiDispatchMsg) {
+ AppendMessageResult appendMessageResult = handlePropertiesForLmqMsg(preEncodeBuffer, msgInner);
+ if (appendMessageResult != null) {
+ return appendMessageResult;
+ }
+ }
+
+ final int msgLen = preEncodeBuffer.getInt(0);
+ preEncodeBuffer.position(0);
+ preEncodeBuffer.limit(msgLen);
+
// PHY OFFSET
long wroteOffset = fileFromOffset + byteBuffer.position();
@@ -1872,9 +1944,6 @@ public class CommitLog implements Swappable {
break;
}
- ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
- final int msgLen = preEncodeBuffer.getInt(0);
-
// Determines whether there is sufficient free space
if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
this.msgStoreItemMemory.clear();
@@ -1919,6 +1988,11 @@ public class CommitLog implements Swappable {
byteBuffer.put(preEncodeBuffer);
CommitLog.this.getMessageStore().getPerfCounter().endTick("WRITE_MEMORY_TIME_MS");
msgInner.setEncodedBuff(null);
+
+ if (isMultiDispatchMsg) {
+ CommitLog.this.multiDispatch.updateMultiQueueOffset(msgInner);
+ }
+
return new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgIdSupplier,
msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills, messageNum);
}
@@ -2159,6 +2233,10 @@ public class CommitLog implements Swappable {
return flushManager;
}
+ public static boolean isMultiDispatchMsg(MessageExtBrokerInner msg) {
+ return StringUtils.isNoneBlank(msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH)) && !msg.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
+ }
+
private boolean isCloseReadAhead() {
return !MixAll.isWindows() && !defaultMessageStore.getMessageStoreConfig().isDataReadAheadEnable();
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
index 623509c8b..453c9d1dc 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
@@ -27,7 +27,6 @@ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -38,7 +37,7 @@ import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.FileQueueLifeCycle;
-import org.apache.rocketmq.store.queue.MultiDispatch;
+import org.apache.rocketmq.store.queue.MultiDispatchUtils;
import org.apache.rocketmq.store.queue.QueueOffsetOperator;
import org.apache.rocketmq.store.queue.ReferredIterator;
@@ -702,7 +701,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
this.messageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
}
this.messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
- if (MultiDispatch.checkMultiDispatchQueue(this.messageStore.getMessageStoreConfig(), request)) {
+ if (MultiDispatchUtils.checkMultiDispatchQueue(this.messageStore.getMessageStoreConfig(), request)) {
multiDispatchLmqQueue(request, maxRetries);
}
return;
@@ -776,28 +775,6 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
String topicQueueKey = getTopic() + "-" + getQueueId();
long queueOffset = queueOffsetOperator.getQueueOffset(topicQueueKey);
msg.setQueueOffset(queueOffset);
-
-
- // Handling the multi dispatch message. In the context of a light message queue (as defined in RIP-28),
- // light message queues are constructed based on message properties, which requires special handling of queue offset of the light message queue.
- if (!MultiDispatch.isNeedHandleMultiDispatch(this.messageStore.getMessageStoreConfig(), msg.getTopic())) {
- return;
- }
- String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
- if (StringUtils.isBlank(multiDispatchQueue)) {
- return;
- }
- String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
- Long[] queueOffsets = new Long[queues.length];
- for (int i = 0; i < queues.length; i++) {
- if (this.messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queues[i])) {
- String key = MultiDispatch.lmqQueueKey(queues[i]);
- queueOffsets[i] = queueOffsetOperator.getLmqOffset(key);
- }
- }
- MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
- StringUtils.join(queueOffsets, MixAll.MULTI_DISPATCH_QUEUE_SPLITTER));
- msg.removeWaitStorePropertyString();
}
@Override
@@ -805,23 +782,6 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
short messageNum) {
String topicQueueKey = getTopic() + "-" + getQueueId();
queueOffsetOperator.increaseQueueOffset(topicQueueKey, messageNum);
-
- // Handling the multi dispatch message. In the context of a light message queue (as defined in RIP-28),
- // light message queues are constructed based on message properties, which requires special handling of queue offset of the light message queue.
- if (!MultiDispatch.isNeedHandleMultiDispatch(this.messageStore.getMessageStoreConfig(), msg.getTopic())) {
- return;
- }
- String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
- if (StringUtils.isBlank(multiDispatchQueue)) {
- return;
- }
- String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
- for (int i = 0; i < queues.length; i++) {
- if (this.messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queues[i])) {
- String key = MultiDispatch.lmqQueueKey(queues[i]);
- queueOffsetOperator.increaseLmqOffset(key, (short) 1);
- }
- }
}
private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index 99a54e2d7..dc5f312e5 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -2112,7 +2112,6 @@ public class DefaultMessageStore implements MessageStore {
}
}
-
@Override
public void increaseOffset(MessageExtBrokerInner msg, short messageNum) {
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java b/store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java
index c1d808728..20e9a652b 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java
@@ -29,6 +29,7 @@ import org.apache.rocketmq.common.message.MessageVersion;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
public class MessageExtEncoder {
protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
@@ -38,20 +39,22 @@ public class MessageExtEncoder {
// The maximum length of the full message.
private int maxMessageSize;
private final int crc32ReservedLength;
+ private MessageStoreConfig messageStoreConfig;
- public MessageExtEncoder(final int maxMessageBodySize) {
- this(maxMessageBodySize, false);
+ public MessageExtEncoder(final int maxMessageBodySize, final MessageStoreConfig messageStoreConfig) {
+ this(messageStoreConfig);
}
- public MessageExtEncoder(final int maxMessageBodySize, boolean enabledAppendPropCRC) {
+ public MessageExtEncoder(final MessageStoreConfig messageStoreConfig) {
ByteBufAllocator alloc = UnpooledByteBufAllocator.DEFAULT;
+ this.messageStoreConfig = messageStoreConfig;
+ this.maxMessageBodySize = messageStoreConfig.getMaxMessageSize();
//Reserve 64kb for encoding buffer outside body
int maxMessageSize = Integer.MAX_VALUE - maxMessageBodySize >= 64 * 1024 ?
maxMessageBodySize + 64 * 1024 : Integer.MAX_VALUE;
byteBuf = alloc.directBuffer(maxMessageSize);
- this.maxMessageBodySize = maxMessageBodySize;
this.maxMessageSize = maxMessageSize;
- this.crc32ReservedLength = enabledAppendPropCRC ? CommitLog.CRC32_RESERVED_LEN : 0;
+ this.crc32ReservedLength = messageStoreConfig.isEnabledAppendPropCRC() ? CommitLog.CRC32_RESERVED_LEN : 0;
}
public static int calMsgLength(MessageVersion messageVersion,
@@ -79,8 +82,103 @@ public class MessageExtEncoder {
+ 2 + (Math.max(propertiesLength, 0)); //propertiesLength
}
+ public static int calMsgLengthNoProperties(MessageVersion messageVersion,
+ int sysFlag, int bodyLength, int topicLength) {
+
+ int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+ int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
+
+ return 4 //TOTALSIZE
+ + 4 //MAGICCODE
+ + 4 //BODYCRC
+ + 4 //QUEUEID
+ + 4 //FLAG
+ + 8 //QUEUEOFFSET
+ + 8 //PHYSICALOFFSET
+ + 4 //SYSFLAG
+ + 8 //BORNTIMESTAMP
+ + bornhostLength //BORNHOST
+ + 8 //STORETIMESTAMP
+ + storehostAddressLength //STOREHOSTADDRESS
+ + 4 //RECONSUMETIMES
+ + 8 //Prepared Transaction Offset
+ + 4 + (Math.max(bodyLength, 0)) //BODY
+ + messageVersion.getTopicLengthSize() + topicLength; //TOPIC
+ }
+
+ public PutMessageResult encodeWithoutProperties(MessageExtBrokerInner msgInner) {
+
+ final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
+ final int topicLength = topicData.length;
+
+ final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
+
+ // Exceeds the maximum message body
+ if (bodyLength > this.maxMessageBodySize) {
+ CommitLog.log.warn("message body size exceeded, msg body size: " + bodyLength
+ + ", maxMessageSize: " + this.maxMessageBodySize);
+ return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
+ }
+
+ final int msgLenNoProperties = calMsgLengthNoProperties(msgInner.getVersion(), msgInner.getSysFlag(), bodyLength, topicLength);
+
+ // 1 TOTALSIZE
+ this.byteBuf.writeInt(msgLenNoProperties);
+ // 2 MAGICCODE
+ this.byteBuf.writeInt(msgInner.getVersion().getMagicCode());
+ // 3 BODYCRC
+ this.byteBuf.writeInt(msgInner.getBodyCRC());
+ // 4 QUEUEID
+ this.byteBuf.writeInt(msgInner.getQueueId());
+ // 5 FLAG
+ this.byteBuf.writeInt(msgInner.getFlag());
+ // 6 QUEUEOFFSET, need update later
+ this.byteBuf.writeLong(0);
+ // 7 PHYSICALOFFSET, need update later
+ this.byteBuf.writeLong(0);
+ // 8 SYSFLAG
+ this.byteBuf.writeInt(msgInner.getSysFlag());
+ // 9 BORNTIMESTAMP
+ this.byteBuf.writeLong(msgInner.getBornTimestamp());
+
+ // 10 BORNHOST
+ ByteBuffer bornHostBytes = msgInner.getBornHostBytes();
+ this.byteBuf.writeBytes(bornHostBytes.array());
+
+ // 11 STORETIMESTAMP
+ this.byteBuf.writeLong(msgInner.getStoreTimestamp());
+
+ // 12 STOREHOSTADDRESS
+ ByteBuffer storeHostBytes = msgInner.getStoreHostBytes();
+ this.byteBuf.writeBytes(storeHostBytes.array());
+
+ // 13 RECONSUMETIMES
+ this.byteBuf.writeInt(msgInner.getReconsumeTimes());
+ // 14 Prepared Transaction Offset
+ this.byteBuf.writeLong(msgInner.getPreparedTransactionOffset());
+ // 15 BODY
+ this.byteBuf.writeInt(bodyLength);
+ if (bodyLength > 0)
+ this.byteBuf.writeBytes(msgInner.getBody());
+
+ // 16 TOPIC
+ if (MessageVersion.MESSAGE_VERSION_V2.equals(msgInner.getVersion())) {
+ this.byteBuf.writeShort((short) topicLength);
+ } else {
+ this.byteBuf.writeByte((byte) topicLength);
+ }
+ this.byteBuf.writeBytes(topicData);
+
+ return null;
+ }
+
public PutMessageResult encode(MessageExtBrokerInner msgInner) {
this.byteBuf.clear();
+
+ if (messageStoreConfig.isEnableMultiDispatch() && CommitLog.isMultiDispatchMsg(msgInner)) {
+ return encodeWithoutProperties(msgInner);
+ }
+
/**
* Serialize message
*/
@@ -303,7 +401,7 @@ public class MessageExtEncoder {
}
public ByteBuffer getEncoderBuffer() {
- return this.byteBuf.nioBuffer();
+ return this.byteBuf.nioBuffer(0, this.byteBuf.capacity());
}
public int getMaxMessageBodySize() {
@@ -322,12 +420,8 @@ public class MessageExtEncoder {
private final MessageExtEncoder encoder;
private final StringBuilder keyBuilder;
- PutMessageThreadLocal(int size) {
- this(size, false);
- }
-
- PutMessageThreadLocal(int size, boolean enabledAppendPropCRC) {
- encoder = new MessageExtEncoder(size, enabledAppendPropCRC);
+ PutMessageThreadLocal(MessageStoreConfig messageStoreConfig) {
+ encoder = new MessageExtEncoder(messageStoreConfig);
keyBuilder = new StringBuilder();
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/MultiDispatch.java b/store/src/main/java/org/apache/rocketmq/store/MultiDispatch.java
new file mode 100644
index 000000000..5bc587a8e
--- /dev/null
+++ b/store/src/main/java/org/apache/rocketmq/store/MultiDispatch.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.store;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
+
+/**
+ * MultiDispatch for lmq, not-thread-safe
+ */
+public class MultiDispatch {
+ private final StringBuilder keyBuilder = new StringBuilder();
+ private final DefaultMessageStore messageStore;
+ private static final short VALUE_OF_EACH_INCREMENT = 1;
+
+ public MultiDispatch(DefaultMessageStore messageStore) {
+ this.messageStore = messageStore;
+ }
+
+ public String queueKey(String queueName, MessageExtBrokerInner msgInner) {
+ keyBuilder.delete(0, keyBuilder.length());
+ keyBuilder.append(queueName);
+ keyBuilder.append('-');
+ int queueId = msgInner.getQueueId();
+ if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
+ queueId = 0;
+ }
+ keyBuilder.append(queueId);
+ return keyBuilder.toString();
+ }
+
+ public void wrapMultiDispatch(final MessageExtBrokerInner msg) {
+
+ String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+ String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+ Long[] queueOffsets = new Long[queues.length];
+ if (messageStore.getMessageStoreConfig().isEnableLmq()) {
+ for (int i = 0; i < queues.length; i++) {
+ String key = queueKey(queues[i], msg);
+ if (MixAll.isLmq(key)) {
+ queueOffsets[i] = messageStore.getQueueStore().getLmqQueueOffset(key);
+ }
+ }
+ }
+ MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
+ StringUtils.join(queueOffsets, MixAll.MULTI_DISPATCH_QUEUE_SPLITTER));
+ msg.removeWaitStorePropertyString();
+ }
+
+ public void updateMultiQueueOffset(final MessageExtBrokerInner msgInner) {
+ String multiDispatchQueue = msgInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+ String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+ for (String queue : queues) {
+ String key = queueKey(queue, msgInner);
+ if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(key)) {
+ messageStore.getQueueStore().increaseLmqOffset(key, VALUE_OF_EACH_INCREMENT);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/AbstractConsumeQueueStore.java b/store/src/main/java/org/apache/rocketmq/store/queue/AbstractConsumeQueueStore.java
index 30054fa50..d76b05577 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/AbstractConsumeQueueStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/AbstractConsumeQueueStore.java
@@ -74,6 +74,16 @@ public abstract class AbstractConsumeQueueStore implements ConsumeQueueStoreInte
consumeQueue.increaseQueueOffset(this.queueOffsetOperator, msg, messageNum);
}
+ @Override
+ public void increaseLmqOffset(String queueKey, short messageNum) {
+ queueOffsetOperator.increaseLmqOffset(queueKey, messageNum);
+ }
+
+ @Override
+ public long getLmqQueueOffset(String queueKey) {
+ return queueOffsetOperator.getLmqOffset(queueKey);
+ }
+
@Override
public void removeTopicQueueTable(String topic, Integer queueId) {
this.queueOffsetOperator.remove(topic, queueId);
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java b/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java
index c65f2a68b..768c782b1 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java
@@ -181,7 +181,6 @@ public interface ConsumeQueueInterface extends FileQueueLifeCycle {
*/
void assignQueueOffset(QueueOffsetOperator queueOffsetAssigner, MessageExtBrokerInner msg) throws RocksDBException;
-
/**
* Increase queue offset.
* @param queueOffsetAssigner the delegated queue offset assigner
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStoreInterface.java b/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStoreInterface.java
index 268803dcc..e68880a82 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStoreInterface.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStoreInterface.java
@@ -183,6 +183,20 @@ public interface ConsumeQueueStoreInterface {
*/
void increaseQueueOffset(MessageExtBrokerInner msg, short messageNum);
+ /**
+ * Increase lmq offset
+ * @param queueKey
+ * @param messageNum
+ */
+ void increaseLmqOffset(String queueKey, short messageNum);
+
+ /**
+ * get lmq queue offset
+ * @param queueKey
+ * @return
+ */
+ long getLmqQueueOffset(String queueKey);
+
/**
* recover topicQueue table by minPhyOffset
* @param minPhyOffset
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatch.java b/store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatchUtils.java
similarity index 78%
rename from store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatch.java
rename to store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatchUtils.java
index d6291d908..44397a2fc 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatch.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/MultiDispatchUtils.java
@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.store.queue;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
@@ -27,7 +25,7 @@ import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.config.MessageStoreConfig;
-public class MultiDispatch {
+public class MultiDispatchUtils {
public static String lmqQueueKey(String queueName) {
StringBuilder keyBuilder = new StringBuilder();
@@ -60,17 +58,4 @@ public class MultiDispatch {
}
return true;
}
-
- public static List<DispatchRequest> checkMultiDispatchQueue(MessageStoreConfig messageStoreConfig, List<DispatchRequest> dispatchRequests) {
- if (!messageStoreConfig.isEnableMultiDispatch() || dispatchRequests == null || dispatchRequests.size() == 0) {
- return null;
- }
- List<DispatchRequest> result = new ArrayList<>();
- for (DispatchRequest dispatchRequest : dispatchRequests) {
- if (checkMultiDispatchQueue(messageStoreConfig, dispatchRequest)) {
- result.add(dispatchRequest);
- }
- }
- return dispatchRequests;
- }
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java b/store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java
index 8da374828..5b4bf994e 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java
@@ -71,9 +71,9 @@ public class QueueOffsetOperator {
return this.lmqTopicQueueTable.get(topicQueueKey);
}
- public void increaseLmqOffset(String topicQueueKey, short messageNum) {
- Long lmqOffset = ConcurrentHashMapUtils.computeIfAbsent(this.lmqTopicQueueTable, topicQueueKey, k -> 0L);
- this.lmqTopicQueueTable.put(topicQueueKey, lmqOffset + messageNum);
+ public void increaseLmqOffset(String queueKey, short messageNum) {
+ Long lmqOffset = ConcurrentHashMapUtils.computeIfAbsent(this.lmqTopicQueueTable, queueKey, k -> 0L);
+ this.lmqTopicQueueTable.put(queueKey, lmqOffset + messageNum);
}
public long currentQueueOffset(String topicQueueKey) {
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java
index 759be395d..5a981bb4d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java
@@ -19,14 +19,10 @@ package org.apache.rocketmq.store.queue;
import java.nio.ByteBuffer;
import java.util.List;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.BoundaryType;
-import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.common.message.MessageAccessor;
-import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -217,50 +213,12 @@ public class RocksDBConsumeQueue implements ConsumeQueueInterface {
queueOffsetOperator.updateQueueOffset(topicQueueKey, queueOffset);
}
msg.setQueueOffset(queueOffset);
-
- // Handling the multi dispatch message. In the context of a light message queue (as defined in RIP-28),
- // light message queues are constructed based on message properties, which requires special handling of queue offset of the light message queue.
- if (!MultiDispatch.isNeedHandleMultiDispatch(this.messageStore.getMessageStoreConfig(), msg.getTopic())) {
- return;
- }
- String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
- if (StringUtils.isBlank(multiDispatchQueue)) {
- return;
- }
- String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
- Long[] queueOffsets = new Long[queues.length];
- for (int i = 0; i < queues.length; i++) {
- if (this.messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queues[i])) {
- String key = MultiDispatch.lmqQueueKey(queues[i]);
- queueOffsets[i] = queueOffsetOperator.getLmqTopicQueueNextOffset(key);
- }
- }
- MessageAccessor.putProperty(msg, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
- StringUtils.join(queueOffsets, MixAll.MULTI_DISPATCH_QUEUE_SPLITTER));
- msg.removeWaitStorePropertyString();
}
@Override
public void increaseQueueOffset(QueueOffsetOperator queueOffsetOperator, MessageExtBrokerInner msg, short messageNum) {
String topicQueueKey = getTopic() + "-" + getQueueId();
queueOffsetOperator.increaseQueueOffset(topicQueueKey, messageNum);
-
- // Handling the multi dispatch message. In the context of a light message queue (as defined in RIP-28),
- // light message queues are constructed based on message properties, which requires special handling of queue offset of the light message queue.
- if (!MultiDispatch.isNeedHandleMultiDispatch(this.messageStore.getMessageStoreConfig(), msg.getTopic())) {
- return;
- }
- String multiDispatchQueue = msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
- if (StringUtils.isBlank(multiDispatchQueue)) {
- return;
- }
- String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
- for (int i = 0; i < queues.length; i++) {
- if (this.messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queues[i])) {
- String key = MultiDispatch.lmqQueueKey(queues[i]);
- queueOffsetOperator.increaseLmqOffset(key, (short) 1);
- }
- }
}
@Override
diff --git a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
index 87bfe85da..374857149 100644
--- a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
@@ -44,7 +44,7 @@ public class AppendCallbackTest {
AppendMessageCallback callback;
- MessageExtEncoder batchEncoder = new MessageExtEncoder(10 * 1024 * 1024);
+ MessageExtEncoder batchEncoder;
@Before
public void init() throws Exception {
@@ -53,12 +53,14 @@ public class AppendCallbackTest {
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
messageStoreConfig.setMaxHashSlotNum(100);
messageStoreConfig.setMaxIndexNum(100 * 10);
+ messageStoreConfig.setMaxMessageSize(10 * 1024 * 1024);
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore" + File.separator + "commitlog");
//too much reference
DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, new BrokerConfig(), new ConcurrentHashMap<>());
CommitLog commitLog = new CommitLog(messageStore);
- callback = commitLog.new DefaultAppendMessageCallback();
+ callback = commitLog.new DefaultAppendMessageCallback(messageStoreConfig);
+ batchEncoder = new MessageExtEncoder(messageStoreConfig);
}
@After
diff --git a/store/src/test/java/org/apache/rocketmq/store/AppendPropCRCTest.java b/store/src/test/java/org/apache/rocketmq/store/AppendPropCRCTest.java
index c8ed4d74d..d882fc9d9 100644
--- a/store/src/test/java/org/apache/rocketmq/store/AppendPropCRCTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/AppendPropCRCTest.java
@@ -56,6 +56,7 @@ public class AppendPropCRCTest {
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
messageStoreConfig.setMaxHashSlotNum(100);
messageStoreConfig.setMaxIndexNum(100 * 10);
+ messageStoreConfig.setMaxMessageSize(10 * 1024 * 1024);
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore" + File.separator + "commitlog");
messageStoreConfig.setForceVerifyPropCRC(true);
@@ -63,8 +64,8 @@ public class AppendPropCRCTest {
//too much reference
DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, new BrokerConfig(), new ConcurrentHashMap<>());
commitLog = new CommitLog(messageStore);
- encoder = new MessageExtEncoder(10 * 1024 * 1024, true);
- callback = commitLog.new DefaultAppendMessageCallback();
+ encoder = new MessageExtEncoder(messageStoreConfig);
+ callback = commitLog.new DefaultAppendMessageCallback(messageStoreConfig);
}
@After
diff --git a/store/src/test/java/org/apache/rocketmq/store/MultiDispatchTest.java b/store/src/test/java/org/apache/rocketmq/store/MultiDispatchTest.java
index 2447bbf68..eae5eaa07 100644
--- a/store/src/test/java/org/apache/rocketmq/store/MultiDispatchTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/MultiDispatchTest.java
@@ -28,20 +28,19 @@ import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.config.MessageStoreConfig;
-import org.apache.rocketmq.store.queue.MultiDispatch;
+import org.apache.rocketmq.store.queue.MultiDispatchUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.rocksdb.RocksDBException;
-import static org.apache.rocketmq.store.config.StorePathConfigHelper.getStorePathConsumeQueue;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class MultiDispatchTest {
- private ConsumeQueue consumeQueue;
+ private MultiDispatch multiDispatch;
private DefaultMessageStore messageStore;
@@ -61,8 +60,7 @@ public class MultiDispatchTest {
BrokerConfig brokerConfig = new BrokerConfig();
//too much reference
messageStore = new DefaultMessageStore(messageStoreConfig, null, null, brokerConfig, new ConcurrentHashMap<>());
- consumeQueue = new ConsumeQueue("xxx", 0,
- getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir()), messageStoreConfig.getMappedFileSizeConsumeQueue(), messageStore);
+ multiDispatch = new MultiDispatch(messageStore);
}
@After
@@ -74,14 +72,14 @@ public class MultiDispatchTest {
public void lmqQueueKey() {
MessageExtBrokerInner messageExtBrokerInner = mock(MessageExtBrokerInner.class);
when(messageExtBrokerInner.getQueueId()).thenReturn(2);
- String ret = MultiDispatch.lmqQueueKey("%LMQ%lmq123");
+ String ret = MultiDispatchUtils.lmqQueueKey("%LMQ%lmq123");
assertEquals(ret, "%LMQ%lmq123-0");
}
@Test
public void wrapMultiDispatch() throws RocksDBException {
MessageExtBrokerInner messageExtBrokerInner = buildMessageMultiQueue();
- messageStore.assignOffset(messageExtBrokerInner);
+ multiDispatch.wrapMultiDispatch(messageExtBrokerInner);
assertEquals(messageExtBrokerInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET), "0,0");
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/kv/CompactionLogTest.java b/store/src/test/java/org/apache/rocketmq/store/kv/CompactionLogTest.java
index df3c31c6e..e113b18f1 100644
--- a/store/src/test/java/org/apache/rocketmq/store/kv/CompactionLogTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/kv/CompactionLogTest.java
@@ -86,7 +86,7 @@ public class CompactionLogTest {
int compactionCqFileSize = 1024;
- private static MessageExtEncoder encoder = new MessageExtEncoder(1024);
+ private static MessageExtEncoder encoder = new MessageExtEncoder(1024, new MessageStoreConfig());
private static SocketAddress storeHost;
private static SocketAddress bornHost;
--
2.32.0.windows.2
From 70dc93abbcb9bf161378d66fcaca55bedc78b905 Mon Sep 17 00:00:00 2001
From: yangguodong <1174533476@qq.com>
Date: Wed, 8 Nov 2023 21:14:54 -0600
Subject: [PATCH 2/2] Fix tiered store README.md error about Configuration
(#7436)
* Fix tiered store README.md error about Configuration
* Fix change tieredStoreFilePath to tieredStoreFilepath
* revert README.md change
---------
Co-authored-by: yangguodong.cn <yangguodong.cn@bytedance.com>
---
.../tieredstore/common/TieredMessageStoreConfig.java | 10 +++++-----
.../tieredstore/provider/posix/PosixFileSegment.java | 4 ++--
.../rocketmq/tieredstore/file/TieredCommitLogTest.java | 2 +-
.../provider/posix/PosixFileSegmentTest.java | 2 +-
4 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/common/TieredMessageStoreConfig.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/common/TieredMessageStoreConfig.java
index 595db6b86..a112ea6b1 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/common/TieredMessageStoreConfig.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/common/TieredMessageStoreConfig.java
@@ -115,7 +115,7 @@ public class TieredMessageStoreConfig {
private long readAheadCacheExpireDuration = 10 * 1000;
private double readAheadCacheSizeThresholdRate = 0.3;
- private String tieredStoreFilePath = "";
+ private String tieredStoreFilepath = "";
private String objectStoreEndpoint = "";
@@ -350,12 +350,12 @@ public class TieredMessageStoreConfig {
this.readAheadCacheSizeThresholdRate = rate;
}
- public String getTieredStoreFilePath() {
- return tieredStoreFilePath;
+ public String getTieredStoreFilepath() {
+ return tieredStoreFilepath;
}
- public void setTieredStoreFilePath(String tieredStoreFilePath) {
- this.tieredStoreFilePath = tieredStoreFilePath;
+ public void setTieredStoreFilepath(String tieredStoreFilepath) {
+ this.tieredStoreFilepath = tieredStoreFilepath;
}
public void setObjectStoreEndpoint(String objectStoreEndpoint) {
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegment.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegment.java
index 7e949cb28..708ce33f9 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegment.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegment.java
@@ -66,8 +66,8 @@ public class PosixFileSegment extends TieredFileSegment {
super(storeConfig, fileType, filePath, baseOffset);
// basePath
- String basePath = StringUtils.defaultString(storeConfig.getTieredStoreFilePath(),
- StringUtils.appendIfMissing(storeConfig.getTieredStoreFilePath(), File.separator));
+ String basePath = StringUtils.defaultString(storeConfig.getTieredStoreFilepath(),
+ StringUtils.appendIfMissing(storeConfig.getTieredStoreFilepath(), File.separator));
// fullPath: basePath/hash_cluster/broker/topic/queueId/fileType/baseOffset
String brokerClusterName = storeConfig.getBrokerClusterName();
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredCommitLogTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredCommitLogTest.java
index 6693d3cb7..80cdba977 100644
--- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredCommitLogTest.java
+++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredCommitLogTest.java
@@ -49,7 +49,7 @@ public class TieredCommitLogTest {
TieredMessageStoreConfig storeConfig = new TieredMessageStoreConfig();
storeConfig.setBrokerName("brokerName");
storeConfig.setStorePathRootDir(storePath);
- storeConfig.setTieredStoreFilePath(storePath + File.separator);
+ storeConfig.setTieredStoreFilepath(storePath + File.separator);
storeConfig.setTieredBackendServiceProvider("org.apache.rocketmq.tieredstore.provider.posix.PosixFileSegment");
storeConfig.setCommitLogRollingInterval(0);
storeConfig.setTieredStoreCommitLogMaxSize(1000);
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegmentTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegmentTest.java
index db33ae847..ede62b8ce 100644
--- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegmentTest.java
+++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/posix/PosixFileSegmentTest.java
@@ -42,7 +42,7 @@ public class PosixFileSegmentTest {
@Before
public void setUp() {
storeConfig = new TieredMessageStoreConfig();
- storeConfig.setTieredStoreFilePath(storePath);
+ storeConfig.setTieredStoreFilepath(storePath);
mq = new MessageQueue("OSSFileSegmentTest", "broker", 0);
TieredStoreExecutor.init();
}
--
2.32.0.windows.2
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。
马建仓 AI 助手