diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java index 7f861e7ce0..b31c71e06d 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java @@ -54,7 +54,9 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; import org.apache.rocketmq.remoting.netty.RemotingResponseCallback; import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.store.DefaultMessageStore; import org.apache.rocketmq.store.MessageExtBrokerInner; +import org.apache.rocketmq.store.MessageStore; import org.apache.rocketmq.store.PutMessageResult; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.apache.rocketmq.store.config.StorePathConfigHelper; @@ -644,7 +646,13 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement private String diskUtil() { double physicRatio = 100; - String storePath = this.brokerController.getMessageStoreConfig().getStorePathCommitLog(); + String storePath; + MessageStore messageStore = this.brokerController.getMessageStore(); + if (messageStore instanceof DefaultMessageStore) { + storePath = ((DefaultMessageStore) messageStore).getStorePathPhysic(); + } else { + storePath = this.brokerController.getMessageStoreConfig().getStorePathCommitLog(); + } String[] paths = storePath.trim().split(MessageStoreConfig.MULTI_PATH_SPLITTER); for (String storePathPhysic : paths) { physicRatio = Math.min(physicRatio, UtilAll.getDiskPartitionSpaceUsedPercent(storePathPhysic)); 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 09e153494d..ed882aa681 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java @@ -1672,7 +1672,7 @@ public class DefaultMessageStore implements MessageStore { cleanImmediately = false; { - String commitLogStorePath = DefaultMessageStore.this.getMessageStoreConfig().getStorePathCommitLog(); + String commitLogStorePath = DefaultMessageStore.this.getStorePathPhysic(); String[] storePaths = commitLogStorePath.trim().split(MessageStoreConfig.MULTI_PATH_SPLITTER); Set fullStorePath = new HashSet<>(); double minPhysicRatio = 100; @@ -1714,8 +1714,7 @@ public class DefaultMessageStore implements MessageStore { } { - String storePathLogics = StorePathConfigHelper - .getStorePathConsumeQueue(DefaultMessageStore.this.getMessageStoreConfig().getStorePathRootDir()); + String storePathLogics = DefaultMessageStore.this.getStorePathLogic(); double logicsRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathLogics); if (logicsRatio > diskSpaceWarningLevelRatio) { boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskFull();