public void popMessageAsync(
        final String brokerName, final String addr, final PopMessageRequestHeader requestHeader,
        final long timeoutMillis, final PopCallback popCallback
    ) throws RemotingException, InterruptedException {
        final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader);
        this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
            @Override
            public void operationComplete(ResponseFuture responseFuture) {

            }

            @Override
            public void operationSucceed(RemotingCommand response) {
                try {
                    PopResult popResult = MQClientAPIImpl.this.processPopResponse(brokerName, response, requestHeader.getTopic(), requestHeader);
                    popCallback.onSuccess(popResult);
                } catch (Exception e) {
                    popCallback.onException(e);
                }
            }
            @Override
            public void operationFail(Throwable throwable) {
                popCallback.onException(throwable);
            }
        });
    }
@Test
    public void testPopLmqMessage_async() throws Exception {
        final long popTime = System.currentTimeMillis();
        final int invisibleTime = 10 * 1000;
        final String lmqTopic = MixAll.LMQ_PREFIX + "lmq1";
        doAnswer((Answer<Void>) mock -> {
            InvokeCallback callback = mock.getArgument(3);
            RemotingCommand request = mock.getArgument(1);
            ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
            RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
            response.setCode(ResponseCode.SUCCESS);
            response.setOpaque(request.getOpaque());

            PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
            responseHeader.setInvisibleTime(invisibleTime);
            responseHeader.setPopTime(popTime);
            responseHeader.setReviveQid(0);
            responseHeader.setRestNum(1);
            StringBuilder startOffsetInfo = new StringBuilder(64);
            ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
            responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
            StringBuilder msgOffsetInfo = new StringBuilder(64);
            ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
            responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
            response.setRemark("FOUND");
            response.makeCustomHeaderToNet();

            MessageExt message = new MessageExt();
            message.setQueueId(3);
            message.setFlag(0);
            message.setQueueOffset(5L);
            message.setCommitLogOffset(11111L);
            message.setSysFlag(0);
            message.setBornTimestamp(System.currentTimeMillis());
            message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
            message.setStoreTimestamp(System.currentTimeMillis());
            message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
            message.setBody("body".getBytes());
            message.setTopic(topic);
            message.putUserProperty("key", "value");
            message.putUserProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH, lmqTopic);
            message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, String.valueOf(0));
            response.setBody(MessageDecoder.encode(message, false));
            responseFuture.setResponseCommand(response);
            callback.operationSucceed(responseFuture.getResponseCommand());
            return null;
        }).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
        final CountDownLatch done = new CountDownLatch(1);
        final PopMessageRequestHeader requestHeader = new PopMessageRequestHeader();
        requestHeader.setTopic(lmqTopic);
        mqClientAPI.popMessageAsync(brokerName, brokerAddr, requestHeader, 10 * 1000, new PopCallback() {
            @Override
            public void onSuccess(PopResult popResult) {
                assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.FOUND);
                assertThat(popResult.getRestNum()).isEqualTo(1);
                assertThat(popResult.getInvisibleTime()).isEqualTo(invisibleTime);
                assertThat(popResult.getPopTime()).isEqualTo(popTime);
                assertThat(popResult.getMsgFoundList()).size().isEqualTo(1);
                assertThat(popResult.getMsgFoundList().get(0).getTopic()).isEqualTo(lmqTopic);
                assertThat(popResult.getMsgFoundList().get(0).getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH))
                    .isEqualTo(lmqTopic);
                done.countDown();
            }

            @Override
            public void onException(Throwable e) {
                Assertions.fail("want no exception but got one", e);
                done.countDown();
            }
        });
        done.await();
    }