[Bug] Tiered Store Query Message Async return different view each time

[AYue-94]

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

macos

RocketMQ version

5.2.0

JDK Version

jdk8

Describe the Bug

query message with key in tiered store return different result each time.

broker log:

# query result is ok
2024-02-29 11:01:58 INFO pool-2-thread-8 - MessageFetcher#queryMessageAsync, query result: 32, topic: TopicTest, topicId: 2, key: abcdefg, maxCount: 32, timestamp: 0-1709175717912

# query result is wrong
2024-02-29 11:02:01 INFO pool-2-thread-7 - MessageFetcher#queryMessageAsync, query result: 31, topic: TopicTest, topicId: 2, key: abcdefg, maxCount: 32, timestamp: 0-1709175721000

so client receive first time 32 msgs, and second time 31 msgs.

Steps to Reproduce

1. broker

start broker with tiered store, and tieredStorageLevel = FORCE.

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
namesrvAddr=127.0.0.1:9876
messageStorePlugIn=org.apache.rocketmq.tieredstore.TieredMessageStore
tieredMetadataServiceProvider=org.apache.rocketmq.tieredstore.metadata.TieredMetadataManager
# custom segment use such as aliyunoss
tieredBackendServiceProvider=org.apache.rocketmq.tieredstore.provider.aliyun.AliyunFileSegment
objectStoreAccessKey=xxx
objectStoreSecretKey=xxx
objectStoreBucket=xxx
objectStoreEndpoint=xx

# force read from tiered store
tieredStorageLevel=FORCE

2. client

use org.apache.rocketmq.example.quickstart.Producer produce msgs , topic=TopicTest, key = abcdefg.

3. wait tiered buffer flush

for about 60s later...

4. client query message

DefaultMQProducer producer = new DefaultMQProducer();
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
while (true) {
    QueryResult result = producer.queryMessage("TopicTest", "abcdefg", 32, 0, System.currentTimeMillis());
    System.out.println(result.getMessageList().size());
    if (result.getMessageList().size() < 32) {
        System.out.println("error!!!");
        break;
    }
}

and you will see client log:

32
32
32
31
error!!!

What Did You Expect to See?

query message async from tiered store should return consistent view.

What Did You See Instead?

query message async from tiered store return different view each time.

Additional Context

No response

[AYue-94]

when segment#read0 use different thread, QueryMessageResult#addMessage will encounter thread safety issues.

focus here

https://github.com/apache/rocketmq/blob/develop/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java#L568

[AYue-94]

i will try to fix it

[lizhimins]

Good catch