protected CompletableFuture<Triple<MessageExt, String, Boolean>> getMessageFromRemoteAsync(String topic, long offset, int queueId, String brokerName) {
        try {
            String brokerAddr = this.brokerController.getTopicRouteInfoManager().findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, false);
            if (null == brokerAddr) {
                this.brokerController.getTopicRouteInfoManager().updateTopicRouteInfoFromNameServer(topic, true, false);
                brokerAddr = this.brokerController.getTopicRouteInfoManager().findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, false);

                if (null == brokerAddr) {
                    LOG.warn("can't find broker address for topic {}, {}", topic, brokerName);
                    return CompletableFuture.completedFuture(Triple.of(null, "brokerAddress not found", true)); // maybe offline temporarily, so need retry
                }
            }

            return this.brokerController.getBrokerOuterAPI().pullMessageFromSpecificBrokerAsync(brokerName,
                brokerAddr, this.innerConsumerGroupName, topic, queueId, offset, 1, DEFAULT_PULL_TIMEOUT_MILLIS)
                .thenApply(pullResult -> {
                    if (pullResult.getLeft() != null
                            && PullStatus.FOUND.equals(pullResult.getLeft().getPullStatus())
                            && CollectionUtils.isNotEmpty(pullResult.getLeft().getMsgFoundList())) {
                        return Triple.of(pullResult.getLeft().getMsgFoundList().get(0), "", false);
                    }
                    return Triple.of(null, pullResult.getMiddle(), pullResult.getRight());
                });
        } catch (Exception e) {
            LOG.error("Get message from remote failed. {}, {}, {}, {}", topic, offset, queueId, brokerName, e);
        }

        return CompletableFuture.completedFuture(Triple.of(null, "Get message from remote failed", true)); // need retry
    }
@Test
    public void getMessageFromRemoteAsyncTest_exception_caught() throws Exception {
        when(brokerOuterAPI.pullMessageFromSpecificBrokerAsync(anyString(), anyString(), anyString(), anyString(), anyInt(), anyLong(), anyInt(), anyLong()))
                .thenThrow(new RemotingException("mock remoting exception"));
        Triple<MessageExt, String, Boolean> rst = escapeBridge.getMessageFromRemoteAsync(TEST_TOPIC, 1, DEFAULT_QUEUE_ID, BROKER_NAME).join();
        Assert.assertNull(rst.getLeft());
        Assert.assertEquals("Get message from remote failed", rst.getMiddle());
        Assert.assertTrue(rst.getRight()); // need retry
    }