This is an automated email from the ASF dual-hosted git repository.
lollipopjin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 50ae7c5093 [ISSUE #10462] Add memory backpressure to dispatch (#10470)
50ae7c5093 is described below
commit 50ae7c50930798aced54c4c24467f847eb684aed
Author: lizhimins <[email protected]>
AuthorDate: Thu Jun 11 14:14:11 2026 +0800
[ISSUE #10462] Add memory backpressure to dispatch (#10470)
---
.../rocketmq/tieredstore/MessageStoreConfig.java | 27 +++++++++++++++
.../core/MessageStoreDispatcherImpl.java | 36 +++++++++++--------
.../core/MessageStoreDispatcherImplTest.java | 40 +++++++++++++++++++++-
3 files changed, 88 insertions(+), 15 deletions(-)
diff --git
a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java
b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java
index d22ab80dd8..fcb2e884bf 100644
---
a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java
+++
b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/MessageStoreConfig.java
@@ -122,6 +122,17 @@ public class MessageStoreConfig {
private double readAheadCacheSizeThresholdRate = 0.3;
private int tieredStoreMaxPendingLimit = 10000;
+ /**
+ * Minimum fraction of max heap that must remain free for dispatch to
proceed.
+ * Actual threshold = min(ratio × maxMemory, maxBytes).
+ * Default 0.1 (10%).
+ */
+ private double tieredStoreDispatchMinFreeMemoryRatio = 0.1;
+ /**
+ * Upper cap on the backpressure threshold in bytes, preventing the
ratio-based
+ * threshold from becoming too large on big-heap instances. Default 1GB.
+ */
+ private long tieredStoreDispatchMinFreeMemoryMaxBytes = 1024 * 1024 *
1024L;
private boolean tieredStoreCrcCheckEnable = false;
private String tieredStoreFilePath = "";
@@ -372,6 +383,22 @@ public class MessageStoreConfig {
this.tieredStoreMaxPendingLimit = tieredStoreMaxPendingLimit;
}
+ public double getTieredStoreDispatchMinFreeMemoryRatio() {
+ return tieredStoreDispatchMinFreeMemoryRatio;
+ }
+
+ public void setTieredStoreDispatchMinFreeMemoryRatio(double
tieredStoreDispatchMinFreeMemoryRatio) {
+ this.tieredStoreDispatchMinFreeMemoryRatio =
tieredStoreDispatchMinFreeMemoryRatio;
+ }
+
+ public long getTieredStoreDispatchMinFreeMemoryMaxBytes() {
+ return tieredStoreDispatchMinFreeMemoryMaxBytes;
+ }
+
+ public void setTieredStoreDispatchMinFreeMemoryMaxBytes(long
tieredStoreDispatchMinFreeMemoryMaxBytes) {
+ this.tieredStoreDispatchMinFreeMemoryMaxBytes =
tieredStoreDispatchMinFreeMemoryMaxBytes;
+ }
+
public boolean isTieredStoreCrcCheckEnable() {
return tieredStoreCrcCheckEnable;
}
diff --git
a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java
b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java
index bcc4e225da..52e1c82a1d 100644
---
a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java
+++
b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java
@@ -30,7 +30,6 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ServiceThread;
@@ -69,7 +68,6 @@ public class MessageStoreDispatcherImpl extends ServiceThread
implements Message
protected final FlatFileStore flatFileStore;
protected final MessageStoreExecutor storeExecutor;
protected final MessageStoreFilter topicFilter;
- protected final Semaphore semaphore;
protected final IndexService indexService;
protected final Map<FlatFileInterface, GroupCommitContext>
failedGroupCommitMap;
@@ -78,8 +76,6 @@ public class MessageStoreDispatcherImpl extends ServiceThread
implements Message
this.storeConfig = messageStore.getStoreConfig();
this.defaultStore = messageStore.getDefaultStore();
this.brokerName = storeConfig.getBrokerName();
- this.semaphore = new Semaphore(
- this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
this.topicFilter = messageStore.getTopicFilter();
this.flatFileStore = messageStore.getFlatFileStore();
this.storeExecutor = messageStore.getStoreExecutor();
@@ -97,16 +93,28 @@ public class MessageStoreDispatcherImpl extends
ServiceThread implements Message
return failedGroupCommitMap;
}
- public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ protected boolean isMemoryEnough(FlatFileInterface flatFile) {
+ Runtime runtime = Runtime.getRuntime();
+ long availableMemory = runtime.maxMemory() - runtime.totalMemory() +
runtime.freeMemory();
+ long ratioThreshold = (long) (runtime.maxMemory() *
storeConfig.getTieredStoreDispatchMinFreeMemoryRatio());
+ long memoryThreshold = Math.min(ratioThreshold,
storeConfig.getTieredStoreDispatchMinFreeMemoryMaxBytes());
+ if (availableMemory < memoryThreshold) {
+ log.warn("MessageStore dispatch skipped due to memory
backpressure, " +
+ "availableMemory={}MB, threshold={}MB, topic={}, queueId={}",
+ availableMemory / (1024 * 1024), memoryThreshold / (1024 *
1024),
+ flatFile.getMessageQueue().getTopic(),
flatFile.getMessageQueue().getQueueId());
+ return false;
+ }
+ return true;
+ }
+
+ public void dispatch(FlatFileInterface flatFile) {
+ if (stopped || !isMemoryEnough(flatFile)) {
+ return;
+ }
try {
- if (stopped) {
- return;
- }
- semaphore.acquire();
- this.doScheduleDispatch(flatFile, false)
- .whenComplete((future, throwable) -> semaphore.release());
+ this.doScheduleDispatch(flatFile, false);
} catch (Throwable t) {
- semaphore.release();
log.error("MessageStore dispatch error, topic={}, queueId={}",
flatFile.getMessageQueue().getTopic(),
flatFile.getMessageQueue().getQueueId(), t);
}
@@ -313,7 +321,7 @@ public class MessageStoreDispatcherImpl extends
ServiceThread implements Message
}
}
if (success && repeat) {
- storeExecutor.commonExecutor.submit(() ->
dispatchWithSemaphore(flatFile));
+ storeExecutor.commonExecutor.submit(() ->
dispatch(flatFile));
}
}
);
@@ -378,7 +386,7 @@ public class MessageStoreDispatcherImpl extends
ServiceThread implements Message
log.info("{} service started", this.getServiceName());
while (!this.isStopped()) {
try {
-
flatFileStore.deepCopyFlatFileToList().forEach(this::dispatchWithSemaphore);
+ flatFileStore.deepCopyFlatFileToList().forEach(this::dispatch);
releaseClosedPendingGroupCommit();
diff --git
a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java
b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java
index 7a43e1ede8..2a0b0ad418 100644
---
a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java
+++
b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java
@@ -37,6 +37,7 @@ import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
import org.apache.rocketmq.tieredstore.TieredMessageStore;
import org.apache.rocketmq.tieredstore.common.GroupCommitContext;
import org.apache.rocketmq.tieredstore.file.FlatFileFactory;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
import org.apache.rocketmq.tieredstore.file.FlatFileStore;
import org.apache.rocketmq.tieredstore.file.FlatMessageFile;
import org.apache.rocketmq.tieredstore.index.IndexItem;
@@ -311,9 +312,46 @@ public class MessageStoreDispatcherImplTest {
Mockito.doAnswer(mock -> {
result.set(true);
return true;
- }).when(dispatcherSpy).dispatchWithSemaphore(any());
+ }).when(dispatcherSpy).dispatch(any(FlatFileInterface.class));
dispatcherSpy.start();
Awaitility.await().atMost(Duration.ofSeconds(10)).until(result::get);
dispatcherSpy.shutdown();
}
+
+ @Test
+ public void isMemoryEnoughTest() {
+ MessageStore defaultStore = Mockito.mock(MessageStore.class);
+ messageStore = Mockito.mock(TieredMessageStore.class);
+ IndexService indexService =
+ new IndexStoreService(new FlatFileFactory(metadataStore,
storeConfig), storePath);
+ Mockito.when(messageStore.getDefaultStore()).thenReturn(defaultStore);
+ Mockito.when(messageStore.getStoreConfig()).thenReturn(storeConfig);
+ Mockito.when(messageStore.getStoreExecutor()).thenReturn(executor);
+ Mockito.when(messageStore.getFlatFileStore()).thenReturn(fileStore);
+ Mockito.when(messageStore.getIndexService()).thenReturn(indexService);
+ MessageStoreDispatcherImpl dispatcher = new
MessageStoreDispatcherImpl(messageStore);
+
+ FlatFileInterface flatFile = Mockito.mock(FlatFileInterface.class);
+ Mockito.when(flatFile.getMessageQueue()).thenReturn(mq);
+
+ // (1) memory sufficient: threshold = min(maxMemory * 0, 0) = 0 →
always true
+ storeConfig.setTieredStoreDispatchMinFreeMemoryRatio(0.0f);
+ storeConfig.setTieredStoreDispatchMinFreeMemoryMaxBytes(0L);
+ Assert.assertTrue(dispatcher.isMemoryEnough(flatFile));
+
+ // (2) memory below threshold: threshold = min(maxMemory * 1.0,
MAX_VALUE) = maxMemory → always false
+ storeConfig.setTieredStoreDispatchMinFreeMemoryRatio(1.0f);
+
storeConfig.setTieredStoreDispatchMinFreeMemoryMaxBytes(Long.MAX_VALUE);
+ Assert.assertFalse(dispatcher.isMemoryEnough(flatFile));
+
+ // (3) maxBytes cap: ratio = 1.0 → ratioThreshold = maxMemory
+ // maxBytes = 1 → threshold = min(maxMemory, 1) = 1 → available > 1 →
true
+ storeConfig.setTieredStoreDispatchMinFreeMemoryMaxBytes(1L);
+ Assert.assertTrue(dispatcher.isMemoryEnough(flatFile));
+ // maxBytes = MAX_VALUE → threshold = maxMemory → available <
maxMemory → false
+
storeConfig.setTieredStoreDispatchMinFreeMemoryMaxBytes(Long.MAX_VALUE);
+ Assert.assertFalse(dispatcher.isMemoryEnough(flatFile));
+
+ dispatcher.shutdown();
+ }
}
