(rocketmq) branch develop updated: [ISSUE #10615] Fix missing long-polling notification under CombineConsumeQueue selective double-write (#10616)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lizhimins 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 45254d5e13 [ISSUE #10615] Fix missing long-polling notification under CombineConsumeQueue selective double-write (#10616)
45254d5e13 is described below

commit 45254d5e13e965b8bdece6101ef42fb38472cd89
Author: imzs <im...@foxmail.com>
AuthorDate: Tue Jul 14 13:39:34 2026 +0800

    [ISSUE #10615] Fix missing long-polling notification under CombineConsumeQueue selective double-write (#10616)
---
 .../apache/rocketmq/store/DefaultMessageStore.java | 36 +++++++++++++++-------
 .../apache/rocketmq/store/RocksDBMessageStore.java | 11 +++++++
 .../store/queue/RocksDBConsumeQueueStore.java      |  1 -
 3 files changed, 36 insertions(+), 12 deletions(-)

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 aee767dae2..64ce41e47d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -164,8 +164,6 @@ public class DefaultMessageStore implements MessageStore {
 
     private volatile boolean shutdown = true;
 
-    private boolean notifyMessageArriveInBatch = false;
-
     protected StoreCheckpoint storeCheckpoint;
     private MessageRocksDBStorage messageRocksDBStorage;
     private TimerMessageStore timerMessageStore;
@@ -2746,7 +2744,7 @@ public class DefaultMessageStore implements MessageStore {
                                 currentReputTimestamp = dispatchRequest.getStoreTimestamp();
                                 DefaultMessageStore.this.doDispatch(dispatchRequest);
 
-                                if (!notifyMessageArriveInBatch) {
+                                if (isNotifyMessageArriveWhenReput()) {
                                     notifyMessageArriveIfNecessary(dispatchRequest);
                                 }
 
@@ -3208,6 +3206,30 @@ public class DefaultMessageStore implements MessageStore {
         this.defaultStoreMetricsManager.init(meter, attributesBuilderSupplier, this);
     }
 
+    /**
+     * Decide whether long-polling consumers should be notified during reput.
+     * <p>
+     * Notification is only safe when the consume queue is updated synchronously in the reput dispatch:
+     * <ul>
+     *     <li>a plain file-based {@link ConsumeQueueStore}, which is always written synchronously;</li>
+     *     <li>a {@link CombineConsumeQueueStore} with selective double-write enabled, where the
+     *     file-based store acts as both the assign-offset and read store and is written
+     *     synchronously, while RocksDB CQ is built asynchronously for only part of the topics.</li>
+     * </ul>
+     * In other cases (e.g. reading from RocksDB CQ) the notification is handled by
+     * {@code RocksGroupCommitService} after the CQ is committed, so we can skip it here.
+     */
+    public boolean isNotifyMessageArriveWhenReput() {
+        if (consumeQueueStore instanceof ConsumeQueueStore) {
+            return true;
+        }
+        if (consumeQueueStore instanceof CombineConsumeQueueStore
+            && messageStoreConfig.isRocksdbCQSelectiveDoubleWriteEnable()) {
+            return true;
+        }
+        return false;
+    }
+
     /**
      * Enable transient commitLog store pool only if transientStorePoolEnable is true and broker role is not SLAVE or
      * enableControllerMode is true
@@ -3249,14 +3271,6 @@ public class DefaultMessageStore implements MessageStore {
         return this.messageRocksDBStorage;
     }
 
-    public boolean isNotifyMessageArriveInBatch() {
-        return notifyMessageArriveInBatch;
-    }
-
-    public void setNotifyMessageArriveInBatch(boolean notifyMessageArriveInBatch) {
-        this.notifyMessageArriveInBatch = notifyMessageArriveInBatch;
-    }
-
     public DefaultStoreMetricsManager getDefaultStoreMetricsManager() {
         return defaultStoreMetricsManager;
     }
diff --git a/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java
index 0983dee7f9..4757ac4f68 100644
--- a/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java
@@ -37,4 +37,15 @@ public class RocksDBMessageStore extends DefaultMessageStore {
     public ConsumeQueueStoreInterface createConsumeQueueStore() {
         return new RocksDBConsumeQueueStore(this);
     }
+
+    /**
+     * RocksDB consume queue is built asynchronously by {@code RocksGroupCommitService}, which is
+     * responsible for notifying long-polling consumers once the CQ is committed. Therefore, we must
+     * not notify during reput, otherwise consumers could be woken up before the message is visible
+     * in the consume queue.
+     */
+    @Override
+    public boolean isNotifyMessageArriveWhenReput() {
+        return false;
+    }
 }
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
index 8573ae8147..2ead65ab64 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
@@ -104,7 +104,6 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
      */
     public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
         super(messageStore);
-        messageStore.setNotifyMessageArriveInBatch(true);
 
         String root = messageStoreConfig.getStorePathRootDir();
         File checkFile;