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 #9709] Add enableRunningFlagsInFlush configuration switch for CommitLog

Browse Source 
- Add configuration to control runningFlags usage in CommitLog (default: false)
- Update CommitLog and AllocateMappedFileService to respect the configuration
- Users can enable runningFlags validation as needed

Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
This commit is contained in:
guyinyou
2025-09-19 11:16:18 +08:00
committed by GitHub
parent 0e72809335
commit dbd935fe22
3 changed files with 25 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java
+5 -3
View File
@@ -173,16 +173,18 @@ public class AllocateMappedFileService extends ServiceThread {
MappedFile mappedFile;
boolean writeWithoutMmap = messageStore.getMessageStoreConfig().isWriteWithoutMmap();
RunningFlags runningFlags = messageStore.getMessageStoreConfig().isEnableRunningFlagsInFlush()
? messageStore.getRunningFlags() : null;
if (messageStore.isTransientStorePoolEnable()) {
try {
mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getRunningFlags(), messageStore.getTransientStorePool());
mappedFile.init(req.getFilePath(), req.getFileSize(), runningFlags, messageStore.getTransientStorePool());
} catch (RuntimeException e) {
log.warn("Use default implementation.");
mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize(), messageStore.getRunningFlags(), messageStore.getTransientStorePool(), writeWithoutMmap);
mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize(), runningFlags, messageStore.getTransientStorePool(), writeWithoutMmap);
}
} else {
mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize(), messageStore.getRunningFlags(), writeWithoutMmap);
mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize(), runningFlags, writeWithoutMmap);
}
long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(beginTime);
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+5 -2
View File
@@ -111,15 +111,18 @@ public class CommitLog implements Swappable {
public CommitLog(final DefaultMessageStore messageStore) {
String storePath = messageStore.getMessageStoreConfig().getStorePathCommitLog();
RunningFlags runningFlags = messageStore.getMessageStoreConfig().isEnableRunningFlagsInFlush()
? messageStore.getRunningFlags() : null;
if (storePath.contains(MixAll.MULTI_PATH_SPLITTER)) {
this.mappedFileQueue = new MultiPathMappedFileQueue(messageStore.getMessageStoreConfig(),
messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(),
messageStore.getAllocateMappedFileService(), this::getFullStorePaths, messageStore.getRunningFlags());
messageStore.getAllocateMappedFileService(), this::getFullStorePaths, runningFlags);
} else {
this.mappedFileQueue = new MappedFileQueue(storePath,
messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(),
messageStore.getAllocateMappedFileService(),
messageStore.getRunningFlags(),
runningFlags,
messageStore.getMessageStoreConfig().isWriteWithoutMmap());
}
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+15
View File
@@ -282,6 +282,13 @@ public class MessageStoreConfig {
*/
private boolean autoMessageVersionOnTopicLen = true;
/**
* Whether to use runningFlags when flushing data to disk.
* When disabled, runningFlags will be set to null during MappedFileQueue and MappedFile initialization.
*/
@ImportantField
private boolean enableRunningFlagsInFlush = false;
/**
* It cannot be changed after the broker is started.
* Modifications need to be restarted to take effect.
@@ -2042,4 +2049,12 @@ public class MessageStoreConfig {
public void setEnableAcceleratedRecovery(boolean enableAcceleratedRecovery) {
this.enableAcceleratedRecovery = enableAcceleratedRecovery;
}
public boolean isEnableRunningFlagsInFlush() {
return enableRunningFlagsInFlush;
}
public void setEnableRunningFlagsInFlush(boolean enableRunningFlagsInFlush) {
this.enableRunningFlagsInFlush = enableRunningFlagsInFlush;
}
}