public RemotingCommand viewMessageById(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final ViewMessageRequestHeader requestHeader =
            (ViewMessageRequestHeader) request.decodeCommandCustomHeader(ViewMessageRequestHeader.class);

        response.setOpaque(request.getOpaque());

        final SelectMappedBufferResult selectMappedBufferResult =
            this.brokerController.getMessageStore().selectOneMessageByOffset(requestHeader.getOffset());
        if (selectMappedBufferResult != null) {
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);

            try {
                FileRegion fileRegion =
                    new OneMessageTransfer(response.encodeHeader(selectMappedBufferResult.getSize()),
                        selectMappedBufferResult);
                ctx.channel()
                    .writeAndFlush(fileRegion)
                    .addListener((ChannelFutureListener) future -> {
                        selectMappedBufferResult.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 one message from page cache failed, ", future.cause());
                        }
                    });
            } catch (Throwable e) {
                LOGGER.error("", e);
                selectMappedBufferResult.release();
            }

            return null;
        } else {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("can not find message by the offset, " + requestHeader.getOffset());
        }

        return response;
    }
@Test
    public void testViewMessageById() throws RemotingCommandException {
        ViewMessageRequestHeader viewMessageRequestHeader = new ViewMessageRequestHeader();
        viewMessageRequestHeader.setTopic("topic");
        viewMessageRequestHeader.setOffset(0L);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, viewMessageRequestHeader);
        request.makeCustomHeaderToNet();
        request.setCode(RequestCode.VIEW_MESSAGE_BY_ID);

        when(messageStore.selectOneMessageByOffset(anyLong())).thenReturn(null);
        RemotingCommand response = queryMessageProcessor.processRequest(handlerContext, request);
        Assert.assertEquals(response.getCode(), ResponseCode.SYSTEM_ERROR);

        when(messageStore.selectOneMessageByOffset(anyLong())).thenReturn(new SelectMappedBufferResult(0, null, 0, null));
        response = queryMessageProcessor.processRequest(handlerContext, request);
        Assert.assertNull(response);
    }