@Override
    public CompletableFuture<PopResult> popMessage(ProxyContext ctx, AddressableMessageQueue messageQueue,
        PopMessageRequestHeader requestHeader, long timeoutMillis) {
        requestHeader.setBornTime(System.currentTimeMillis());
        RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader, ctx.getLanguage());
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        SimpleChannel channel = channelManager.createInvocationChannel(ctx);
        InvocationContext invocationContext = new InvocationContext(future);
        channel.registerInvocationContext(request.getOpaque(), invocationContext);
        ChannelHandlerContext simpleChannelHandlerContext = channel.getChannelHandlerContext();
        try {
            RemotingCommand response = brokerController.getPopMessageProcessor().processRequest(simpleChannelHandlerContext, request);
            if (response != null) {
                invocationContext.handle(response);
                channel.eraseInvocationContext(request.getOpaque());
            }
        } catch (Exception e) {
            future.completeExceptionally(e);
            channel.eraseInvocationContext(request.getOpaque());
            log.error("Failed to process popMessage command", e);
        }
        return future.thenApply(r -> {
            PopStatus popStatus;
            List<MessageExt> messageExtList = new ArrayList<>();
            switch (r.getCode()) {
                case ResponseCode.SUCCESS:
                    popStatus = PopStatus.FOUND;
                    ByteBuffer byteBuffer = ByteBuffer.wrap(r.getBody());
                    messageExtList = MessageDecoder.decodesBatch(
                        byteBuffer,
                        true,
                        false,
                        true
                    );
                    break;
                case ResponseCode.POLLING_FULL:
                    popStatus = PopStatus.POLLING_FULL;
                    break;
                case ResponseCode.POLLING_TIMEOUT:
                case ResponseCode.PULL_NOT_FOUND:
                    popStatus = PopStatus.POLLING_NOT_FOUND;
                    break;
                default:
                    throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, r.getRemark());
            }
            PopResult popResult = new PopResult(popStatus, messageExtList);
            PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) r.readCustomHeader();

            if (popStatus == PopStatus.FOUND) {
                Map<String, Long> startOffsetInfo;
                Map<String, List<Long>> msgOffsetInfo;
                Map<String, Integer> orderCountInfo;
                popResult.setInvisibleTime(responseHeader.getInvisibleTime());
                popResult.setPopTime(responseHeader.getPopTime());
                startOffsetInfo = ExtraInfoUtil.parseStartOffsetInfo(responseHeader.getStartOffsetInfo());
                msgOffsetInfo = ExtraInfoUtil.parseMsgOffsetInfo(responseHeader.getMsgOffsetInfo());
                orderCountInfo = ExtraInfoUtil.parseOrderCountInfo(responseHeader.getOrderCountInfo());
                // <topicMark@queueId, msg queueOffset>
                Map<String, List<Long>> sortMap = new HashMap<>(16);
                for (MessageExt messageExt : messageExtList) {
                    // Value of POP_CK is used to determine whether it is a pop retry,
                    // cause topic could be rewritten by broker.
                    String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(),
                        messageExt.getProperty(MessageConst.PROPERTY_POP_CK), messageExt.getQueueId());
                    if (!sortMap.containsKey(key)) {
                        sortMap.put(key, new ArrayList<>(4));
                    }
                    sortMap.get(key).add(messageExt.getQueueOffset());
                }
                Map<String, String> map = new HashMap<>(5);
                for (MessageExt messageExt : messageExtList) {
                    if (startOffsetInfo == null) {
                        // we should set the check point info to extraInfo field , if the command is popMsg
                        // find pop ck offset
                        String key = messageExt.getTopic() + messageExt.getQueueId();
                        if (!map.containsKey(messageExt.getTopic() + messageExt.getQueueId())) {
                            map.put(key, ExtraInfoUtil.buildExtraInfo(messageExt.getQueueOffset(), responseHeader.getPopTime(), responseHeader.getInvisibleTime(), responseHeader.getReviveQid(),
                                messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId()));
                        }
                        messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK, map.get(key) + MessageConst.KEY_SEPARATOR + messageExt.getQueueOffset());
                    } else {
                        if (messageExt.getProperty(MessageConst.PROPERTY_POP_CK) == null) {
                            String key = ExtraInfoUtil.getStartOffsetInfoMapKey(messageExt.getTopic(), messageExt.getQueueId());
                            int index = sortMap.get(key).indexOf(messageExt.getQueueOffset());
                            Long msgQueueOffset = msgOffsetInfo.get(key).get(index);
                            if (msgQueueOffset != messageExt.getQueueOffset()) {
                                log.warn("Queue offset [{}] of msg is strange, not equal to the stored in msg, {}", msgQueueOffset, messageExt);
                            }

                            messageExt.getProperties().put(MessageConst.PROPERTY_POP_CK,
                                ExtraInfoUtil.buildExtraInfo(startOffsetInfo.get(key), responseHeader.getPopTime(), responseHeader.getInvisibleTime(),
                                    responseHeader.getReviveQid(), messageExt.getTopic(), messageQueue.getBrokerName(), messageExt.getQueueId(), msgQueueOffset)
                            );
                            if (requestHeader.isOrder() && orderCountInfo != null) {
                                Integer count = orderCountInfo.get(key);
                                if (count != null && count > 0) {
                                    messageExt.setReconsumeTimes(count);
                                }
                            }
                        }
                    }
                    messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME, k -> String.valueOf(responseHeader.getPopTime()));
                    messageExt.setBrokerName(messageExt.getBrokerName());
                    messageExt.setTopic(messageQueue.getTopic());
                }
            }
            return popResult;
        });
    }
@Test
    public void testPopMessageWriteAndFlush() throws Exception {
        int reviveQueueId = 1;
        long popTime = System.currentTimeMillis();
        long invisibleTime = 3000L;
        long startOffset = 100L;
        long restNum = 0L;
        StringBuilder startOffsetStringBuilder = new StringBuilder();
        StringBuilder messageOffsetStringBuilder = new StringBuilder();
        List<MessageExt> messageExtList = new ArrayList<>();
        List<Long> messageOffsetList = new ArrayList<>();
        MessageExt message1 = buildMessageExt(topic, 0, startOffset);
        messageExtList.add(message1);
        messageOffsetList.add(startOffset);
        byte[] body1 = MessageDecoder.encode(message1, false);
        MessageExt message2 = buildMessageExt(topic, 0, startOffset + 1);
        messageExtList.add(message2);
        messageOffsetList.add(startOffset + 1);
        ExtraInfoUtil.buildStartOffsetInfo(startOffsetStringBuilder, topic, queueId, startOffset);
        ExtraInfoUtil.buildMsgOffsetInfo(messageOffsetStringBuilder, topic, queueId, messageOffsetList);
        byte[] body2 = MessageDecoder.encode(message2, false);
        ByteBuffer byteBuffer1 = ByteBuffer.wrap(body1);
        ByteBuffer byteBuffer2 = ByteBuffer.wrap(body2);
        ByteBuffer b3 = ByteBuffer.allocate(byteBuffer1.limit() + byteBuffer2.limit());
        b3.put(byteBuffer1);
        b3.put(byteBuffer2);
        PopMessageRequestHeader requestHeader = new PopMessageRequestHeader();
        requestHeader.setInvisibleTime(invisibleTime);
        Mockito.when(popMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> {
            boolean first = argument.getCode() == RequestCode.POP_MESSAGE;
            boolean second = argument.readCustomHeader() instanceof PopMessageRequestHeader;
            return first && second;
        }))).thenAnswer(invocation -> {
            SimpleChannelHandlerContext simpleChannelHandlerContext = invocation.getArgument(0);
            RemotingCommand request = invocation.getArgument(1);
            RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
            response.setOpaque(request.getOpaque());
            response.setCode(ResponseCode.SUCCESS);
            response.setBody(b3.array());
            PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
            responseHeader.setStartOffsetInfo(startOffsetStringBuilder.toString());
            responseHeader.setMsgOffsetInfo(messageOffsetStringBuilder.toString());
            responseHeader.setInvisibleTime(requestHeader.getInvisibleTime());
            responseHeader.setPopTime(popTime);
            responseHeader.setRestNum(restNum);
            responseHeader.setReviveQid(reviveQueueId);
            simpleChannelHandlerContext.writeAndFlush(response);
            return null;
        });
        MessageQueue messageQueue = new MessageQueue(topic, brokerName, queueId);
        CompletableFuture<PopResult> future = localMessageService.popMessage(proxyContext, new AddressableMessageQueue(messageQueue, ""), requestHeader, 1000L);
        PopResult popResult = future.get();
        assertThat(popResult.getPopTime()).isEqualTo(popTime);
        assertThat(popResult.getInvisibleTime()).isEqualTo(invisibleTime);
        assertThat(popResult.getPopStatus()).isEqualTo(PopStatus.FOUND);
        assertThat(popResult.getRestNum()).isEqualTo(restNum);
        assertThat(popResult.getMsgFoundList().size()).isEqualTo(messageExtList.size());
        for (int i = 0; i < popResult.getMsgFoundList().size(); i++) {
            assertMessageExt(popResult.getMsgFoundList().get(i), messageExtList.get(i));
        }
    }