public CompletableFuture<Triple<MessageExt, String, Boolean>> getMessageAsync(String topic, long offset, int queueId, String brokerName, boolean deCompressBody) {
        MessageStore messageStore = brokerController.getMessageStoreByBrokerName(brokerName);
        if (messageStore != null) {
            return messageStore.getMessageAsync(innerConsumerGroupName, topic, queueId, offset, 1, null)
                .thenApply(result -> {
                    if (result == null) {
                        LOG.warn("getMessageResult is null , innerConsumerGroupName {}, topic {}, offset {}, queueId {}", innerConsumerGroupName, topic, offset, queueId);
                        return Triple.of(null, "getMessageResult is null", false); // local store, so no retry
                    }
                    List<MessageExt> list = decodeMsgList(result, deCompressBody);
                    if (list == null || list.isEmpty()) {
                        // OFFSET_FOUND_NULL returned by TieredMessageStore indicates exception occurred
                        boolean needRetry = GetMessageStatus.OFFSET_FOUND_NULL.equals(result.getStatus())
                                && messageStore instanceof TieredMessageStore;
                        LOG.warn("Can not get msg , topic {}, offset {}, queueId {}, needRetry {}, result is {}",
                                topic, offset, queueId, needRetry, result);
                        return Triple.of(null, "Can not get msg", needRetry);
                    }
                    return Triple.of(list.get(0), "", false);
                });
        } else {
            return getMessageFromRemoteAsync(topic, offset, queueId, brokerName);
        }
    }
@Test
    public void getMessageAsyncTest_localStore_decodeNothing_DefaultMessageStore() throws Exception {
        when(brokerController.getMessageStoreByBrokerName(any())).thenReturn(defaultMessageStore);
        for (GetMessageStatus status : GetMessageStatus.values()) {
            GetMessageResult getMessageResult = mockGetMessageResult(0, TEST_TOPIC, null);
            getMessageResult.setStatus(status);
            when(defaultMessageStore.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any()))
                    .thenReturn(CompletableFuture.completedFuture(getMessageResult));
            Triple<MessageExt, String, Boolean> rst = escapeBridge.getMessageAsync(TEST_TOPIC, 0, DEFAULT_QUEUE_ID, BROKER_NAME, false).join();
            Assert.assertNull(rst.getLeft());
            Assert.assertEquals("Can not get msg", rst.getMiddle());
            Assert.assertFalse(rst.getRight()); // DefaultMessageStore, no retry
        }
    }