[Bug] Reput notifies multi-queue arrival before the RocksDB LMQ ConsumeQueue is committed #11272

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.

• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

.

RocketMQ version

.

JDK Version

.

Describe the Bug

We introduced CombineConsumeQueueStore and made the LMQ ConsumeQueue (CQ) build
asynchronous in RocksDB. Specifically, the file-version LMQ CQ construction is gated by
MultiDispatchUtils.checkMultiDispatchQueue(config, request).

When rocksdbCQDoubleWriteEnable && combineCQUseRocksdbForLmq both hold, this predicate
returns false, so the file-version LMQ CQ is not built during reput. Instead, the LMQ
CQ is written and committed asynchronously by RocksGroupCommitService. The correct arrival notification for this path is emitted only after the batch commit, via RocksDBConsumeQueueStore#notifyMessageArriveAndClear -> notifyMessageArriveIfNecessary(dp).

#10173 blocked the file-version LMQ CQ construction, but it did not block the LMQ notification on the reput path. DefaultMessageStore#notifyMessageArriveIfNecessary is a shared entry point used by both:

1. the reput path — ReputMessageService#doReput and DispatchService#dispatch;
2. the post-commit path — RocksDBConsumeQueueStore#notifyMessageArriveAndClear.

On the reput path it still unconditionally invokes notifyMessageArrive4MultiQueue(dispatchRequest),
which fires MessageArrivingListener#arriving(...) for every LMQ. So for a
CombineConsumeQueueStore with combineCQUseRocksdbForLmq=true, the LMQ arrival is notified
during reput, i.e. before the RocksDB CQ batch has actually been committed, then the prematurely woken long-polling consumer reads an empty / stale queue.

Steps to Reproduce

.

What Did You Expect to See?

.

What Did You See Instead?

.

Additional Context

.