@Override
    public CompletableFuture<QueryMessageResult> queryMessageAsync(
        String topic, String key, int maxCount, long begin, long end) {

        long topicId;
        try {
            TopicMetadata topicMetadata = metadataStore.getTopic(topic);
            if (topicMetadata == null) {
                log.info("MessageFetcher#queryMessageAsync, topic metadata not found, topic={}", topic);
                return CompletableFuture.completedFuture(new QueryMessageResult());
            }
            topicId = topicMetadata.getTopicId();
        } catch (Exception e) {
            log.error("MessageFetcher#queryMessageAsync, get topic id failed, topic={}", topic, e);
            return CompletableFuture.completedFuture(new QueryMessageResult());
        }

        CompletableFuture<List<IndexItem>> future = indexService.queryAsync(topic, key, maxCount, begin, end);

        return future.thenCompose(indexItemList -> {
            List<CompletableFuture<SelectMappedBufferResult>> futureList = new ArrayList<>(maxCount);
            for (IndexItem indexItem : indexItemList) {
                if (topicId != indexItem.getTopicId()) {
                    continue;
                }
                FlatMessageFile flatFile =
                    flatFileStore.getFlatFile(new MessageQueue(topic, brokerName, indexItem.getQueueId()));
                if (flatFile == null) {
                    continue;
                }
                CompletableFuture<SelectMappedBufferResult> getMessageFuture = flatFile
                    .getCommitLogAsync(indexItem.getOffset(), indexItem.getSize())
                    .thenApply(messageBuffer -> new SelectMappedBufferResult(
                        indexItem.getOffset(), messageBuffer, indexItem.getSize(), null));
                futureList.add(getMessageFuture);
                if (futureList.size() >= maxCount) {
                    break;
                }
            }
            return CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).thenApply(v -> {
                QueryMessageResult result = new QueryMessageResult();
                futureList.forEach(f -> f.thenAccept(result::addMessage));
                return result;
            });
        }).whenComplete((result, throwable) -> {
            if (result != null) {
                log.info("MessageFetcher#queryMessageAsync, " +
                        "query result={}, topic={}, topicId={}, key={}, maxCount={}, timestamp={}-{}",
                    result.getMessageBufferList().size(), topic, topicId, key, maxCount, begin, end);
            }
        });
    }
@Test
    public void testQueryMessageAsync() throws Exception {
        this.getMessageFromTieredStoreTest();
        mq = dispatcherTest.mq;
        messageStore = dispatcherTest.messageStore;
        storeConfig = dispatcherTest.storeConfig;

        QueryMessageResult queryMessageResult = fetcher.queryMessageAsync(
            mq.getTopic(), "uk", 32, 0L, System.currentTimeMillis()).join();
        Assert.assertEquals(32, queryMessageResult.getMessageBufferList().size());

        queryMessageResult = fetcher.queryMessageAsync(
            mq.getTopic(), "uk", 120, 0L, System.currentTimeMillis()).join();
        Assert.assertEquals(100, queryMessageResult.getMessageBufferList().size());
    }