@Override
    public CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int queueId, long queueOffset) {
        FlatMessageFile flatFile = flatFileStore.getFlatFile(new MessageQueue(topic, brokerName, queueId));
        if (flatFile == null) {
            return CompletableFuture.completedFuture(-1L);
        }

        return flatFile.getConsumeQueueAsync(queueOffset)
            .thenComposeAsync(cqItem -> {
                long commitLogOffset = MessageFormatUtil.getCommitLogOffsetFromItem(cqItem);
                int size = MessageFormatUtil.getSizeFromItem(cqItem);
                return flatFile.getCommitLogAsync(commitLogOffset, size);
            }, messageStore.getStoreExecutor().bufferFetchExecutor)
            .thenApply(MessageFormatUtil::getStoreTimeStamp)
            .exceptionally(e -> {
                log.error("MessageStoreFetcherImpl#getMessageStoreTimeStampAsync: " +
                    "get or decode message failed, topic={}, queue={}, offset={}", topic, queueId, queueOffset, e);
                return -1L;
            });
    }
@Test
    public void testGetMessageStoreTimeStampAsync() throws Exception {
        this.getMessageFromTieredStoreTest();
        mq = dispatcherTest.mq;
        messageStore = dispatcherTest.messageStore;
        storeConfig = dispatcherTest.storeConfig;

        long result1 = fetcher.getEarliestMessageTimeAsync(mq.getTopic(), 0).join();
        Assert.assertEquals(-1L, result1);

        long result2 = fetcher.getEarliestMessageTimeAsync(mq.getTopic(), mq.getQueueId()).join();
        Assert.assertEquals(11L, result2);

        long result3 = fetcher.getMessageStoreTimeStampAsync(mq.getTopic(), 0, 100).join();
        Assert.assertEquals(-1L, result3);

        long result4 = fetcher.getMessageStoreTimeStampAsync(mq.getTopic(), mq.getQueueId(), 100).join();
        Assert.assertEquals(11L, result4);

        long result5 = fetcher.getMessageStoreTimeStampAsync(mq.getTopic(), mq.getQueueId(), 120).join();
        Assert.assertEquals(11L, result5);
    }