public RemotingCommand queryMessage(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response =
            RemotingCommand.createResponseCommand(QueryMessageResponseHeader.class);
        final QueryMessageResponseHeader responseHeader =
            (QueryMessageResponseHeader) response.readCustomHeader();
        final QueryMessageRequestHeader requestHeader =
            (QueryMessageRequestHeader) request
                .decodeCommandCustomHeader(QueryMessageRequestHeader.class);

        response.setOpaque(request.getOpaque());

        String isUniqueKey = request.getExtFields().get(MixAll.UNIQUE_MSG_QUERY_FLAG);
        if (isUniqueKey != null && isUniqueKey.equals("true")) {
            requestHeader.setMaxNum(this.brokerController.getMessageStoreConfig().getDefaultQueryMaxNum());
        }

        final QueryMessageResult queryMessageResult =
            this.brokerController.getMessageStore().queryMessage(requestHeader.getTopic(),
                requestHeader.getKey(), requestHeader.getMaxNum(), requestHeader.getBeginTimestamp(),
                requestHeader.getEndTimestamp());
        assert queryMessageResult != null;

        responseHeader.setIndexLastUpdatePhyoffset(queryMessageResult.getIndexLastUpdatePhyoffset());
        responseHeader.setIndexLastUpdateTimestamp(queryMessageResult.getIndexLastUpdateTimestamp());

        if (queryMessageResult.getBufferTotalSize() > 0) {
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);

            try {
                FileRegion fileRegion =
                    new QueryMessageTransfer(response.encodeHeader(queryMessageResult
                        .getBufferTotalSize()), queryMessageResult);
                ctx.channel()
                    .writeAndFlush(fileRegion)
                    .addListener((ChannelFutureListener) future -> {
                        queryMessageResult.release();
                        Attributes attributes = RemotingMetricsManager.newAttributesBuilder()
                            .put(LABEL_REQUEST_CODE, RemotingHelper.getRequestCodeDesc(request.getCode()))
                            .put(LABEL_RESPONSE_CODE, RemotingHelper.getResponseCodeDesc(response.getCode()))
                            .put(LABEL_RESULT, RemotingMetricsManager.getWriteAndFlushResult(future))
                            .build();
                        RemotingMetricsManager.rpcLatency.record(request.getProcessTimer().elapsed(TimeUnit.MILLISECONDS), attributes);
                        if (!future.isSuccess()) {
                            LOGGER.error("transfer query message by page cache failed, ", future.cause());
                        }
                    });
            } catch (Throwable e) {
                LOGGER.error("", e);
                queryMessageResult.release();
            }

            return null;
        }

        response.setCode(ResponseCode.QUERY_NOT_FOUND);
        response.setRemark("can not find message, maybe time range not correct");
        return response;
    }
@Test
    public void testQueryMessage() throws RemotingCommandException {
        QueryMessageResult result = new QueryMessageResult();
        result.setIndexLastUpdateTimestamp(100);
        result.setIndexLastUpdatePhyoffset(0);
        result.addMessage(new SelectMappedBufferResult(0, null, 0, null));

        when(messageStore.queryMessage(anyString(),anyString(),anyInt(),anyLong(),anyLong())).thenReturn(result);
        RemotingCommand request = createQueryMessageRequest("topic", "msgKey", 1, 100, 200,"false");
        request.makeCustomHeaderToNet();
        RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
        Assert.assertEquals(response.getCode(), ResponseCode.QUERY_NOT_FOUND);

        result.addMessage(new SelectMappedBufferResult(0, null, 1, null));
        when(messageStore.queryMessage(anyString(),anyString(),anyInt(),anyLong(),anyLong())).thenReturn(result);
        response = queryMessageProcessor.processRequest(handlerContext, request);
        Assert.assertNull(response);
    }