@Override
    public CompletableFuture<GetMessageResult> getMessageAsync(String group, String topic,
        int queueId, long offset, int maxMsgNums, MessageFilter messageFilter) {

        // for system topic, force reading from local store
        if (topicFilter.filterTopic(topic)) {
            return next.getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter);
        }

        if (fetchFromCurrentStore(topic, queueId, offset, maxMsgNums)) {
            log.trace("GetMessageAsync from current store, " +
                "topic: {}, queue: {}, offset: {}, maxCount: {}", topic, queueId, offset, maxMsgNums);
        } else {
            log.trace("GetMessageAsync from remote store, " +
                "topic: {}, queue: {}, offset: {}, maxCount: {}", topic, queueId, offset, maxMsgNums);
            return next.getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter);
        }

        Stopwatch stopwatch = Stopwatch.createStarted();
        return fetcher
            .getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter)
            .thenApply(result -> {

                Attributes latencyAttributes = TieredStoreMetricsManager.newAttributesBuilder()
                    .put(TieredStoreMetricsConstant.LABEL_OPERATION, TieredStoreMetricsConstant.OPERATION_API_GET_MESSAGE)
                    .put(TieredStoreMetricsConstant.LABEL_TOPIC, topic)
                    .put(TieredStoreMetricsConstant.LABEL_GROUP, group)
                    .build();
                TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);

                if (result.getStatus() == GetMessageStatus.OFFSET_FOUND_NULL ||
                    result.getStatus() == GetMessageStatus.NO_MATCHED_LOGIC_QUEUE) {

                    if (next.checkInStoreByConsumeOffset(topic, queueId, offset)) {
                        TieredStoreMetricsManager.fallbackTotal.add(1, latencyAttributes);
                        log.debug("GetMessageAsync not found, then back to next store, result: {}, " +
                                "topic: {}, queue: {}, queue offset: {}, offset range: {}-{}",
                            result.getStatus(), topic, queueId, offset, result.getMinOffset(), result.getMaxOffset());
                        return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
                    }
                }

                if (result.getStatus() != GetMessageStatus.FOUND &&
                    result.getStatus() != GetMessageStatus.NO_MESSAGE_IN_QUEUE &&
                    result.getStatus() != GetMessageStatus.NO_MATCHED_LOGIC_QUEUE &&
                    result.getStatus() != GetMessageStatus.OFFSET_TOO_SMALL &&
                    result.getStatus() != GetMessageStatus.OFFSET_OVERFLOW_ONE &&
                    result.getStatus() != GetMessageStatus.OFFSET_OVERFLOW_BADLY) {
                    log.warn("GetMessageAsync not found and message is not in next store, result: {}, " +
                            "topic: {}, queue: {}, queue offset: {}, offset range: {}-{}",
                        result.getStatus(), topic, queueId, offset, result.getMinOffset(), result.getMaxOffset());
                }

                if (result.getStatus() == GetMessageStatus.FOUND) {
                    Attributes messagesOutAttributes = TieredStoreMetricsManager.newAttributesBuilder()
                        .put(TieredStoreMetricsConstant.LABEL_TOPIC, topic)
                        .put(TieredStoreMetricsConstant.LABEL_GROUP, group)
                        .build();
                    TieredStoreMetricsManager.messagesOutTotal.add(result.getMessageCount(), messagesOutAttributes);

                    if (next.getStoreStatsService() != null) {
                        next.getStoreStatsService().getGetMessageTransferredMsgCount().add(result.getMessageCount());
                    }
                }

                // Fix min or max offset according next store at last
                long minOffsetInQueue = next.getMinOffsetInQueue(topic, queueId);
                if (minOffsetInQueue >= 0 && minOffsetInQueue < result.getMinOffset()) {
                    result.setMinOffset(minOffsetInQueue);
                }

                // In general, the local cq offset is slightly greater than the commit offset in read message,
                // so there is no need to update the maximum offset to the local cq offset here,
                // otherwise it will cause repeated consumption after next start offset over commit offset.

                if (storeConfig.isRecordGetMessageResult()) {
                    log.info("GetMessageAsync result, {}, group: {}, topic: {}, queueId: {}, offset: {}, count:{}",
                        result, group, topic, queueId, offset, maxMsgNums);
                }

                return result;
            }).exceptionally(e -> {
                log.error("GetMessageAsync from tiered store failed", e);
                return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
            });
    }
@Test
    public void testGetMessageAsync() {
        GetMessageResult expect = new GetMessageResult();
        expect.setStatus(GetMessageStatus.FOUND);
        expect.setMinOffset(100L);
        expect.setMaxOffset(200L);

        // topic filter
        Mockito.when(defaultStore.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any()))
            .thenReturn(CompletableFuture.completedFuture(expect));
        String groupName = "groupName";
        GetMessageResult result = currentStore.getMessage(
            groupName, TopicValidator.SYSTEM_TOPIC_PREFIX, mq.getQueueId(), 100, 0, null);
        Assert.assertSame(expect, result);

        // fetch from default
        Mockito.when(fetcher.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any()))
            .thenReturn(CompletableFuture.completedFuture(expect));

        result = currentStore.getMessage(
            groupName, mq.getTopic(), mq.getQueueId(), 100, 0, null);
        Assert.assertSame(expect, result);

        expect.setStatus(GetMessageStatus.NO_MATCHED_LOGIC_QUEUE);
        Assert.assertSame(expect, currentStore.getMessage(
            groupName, mq.getTopic(), mq.getQueueId(), 0, 0, null));

        expect.setStatus(GetMessageStatus.OFFSET_OVERFLOW_ONE);
        Assert.assertSame(expect, currentStore.getMessage(
            groupName, mq.getTopic(), mq.getQueueId(), 0, 0, null));

        expect.setStatus(GetMessageStatus.OFFSET_OVERFLOW_BADLY);
        Assert.assertSame(expect, currentStore.getMessage(
            groupName, mq.getTopic(), mq.getQueueId(), 0, 0, null));

        expect.setStatus(GetMessageStatus.OFFSET_RESET);
        Assert.assertSame(expect, currentStore.getMessage(
            groupName, mq.getTopic(), mq.getQueueId(), 0, 0, null));
    }