@Override
    public RemotingCommand processRequest(final ChannelHandlerContext ctx,
                                          RemotingCommand request) throws RemotingCommandException {
        return this.processRequest(ctx.channel(), request, true);
    }
@Test
    public void testSingleAck_appendAck() throws RemotingCommandException {
        {
            // buffer addAk OK
            PopBufferMergeService popBufferMergeService = mock(PopBufferMergeService.class);
            when(popBufferMergeService.addAk(anyInt(), any())).thenReturn(true);
            when(popMessageProcessor.getPopBufferMergeService()).thenReturn(popBufferMergeService);

            AckMessageRequestHeader requestHeader = new AckMessageRequestHeader();
            long ackOffset = MIN_OFFSET_IN_QUEUE + 10;
            requestHeader.setTopic(topic);
            requestHeader.setQueueId(0);
            requestHeader.setOffset(ackOffset);
            requestHeader.setConsumerGroup(MixAll.DEFAULT_CONSUMER_GROUP);
            requestHeader.setExtraInfo("64 1666860736757 60000 4 0 broker-a 0 " + ackOffset);
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader);
            request.makeCustomHeaderToNet();
            RemotingCommand response = ackMessageProcessor.processRequest(handlerContext, request);

            assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
        }

        {
            // buffer addAk fail
            PopBufferMergeService popBufferMergeService = mock(PopBufferMergeService.class);
            when(popBufferMergeService.addAk(anyInt(), any())).thenReturn(false);
            when(popMessageProcessor.getPopBufferMergeService()).thenReturn(popBufferMergeService);
            // store putMessage OK
            PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, null);
            when(messageStore.putMessage(any())).thenReturn(putMessageResult);

            AckMessageRequestHeader requestHeader = new AckMessageRequestHeader();
            long ackOffset = MIN_OFFSET_IN_QUEUE + 10;
            requestHeader.setTopic(topic);
            requestHeader.setQueueId(0);
            requestHeader.setOffset(ackOffset);
            requestHeader.setConsumerGroup(MixAll.DEFAULT_CONSUMER_GROUP);
            requestHeader.setExtraInfo("64 1666860736757 60000 4 0 broker-a 0 " + ackOffset);
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader);
            request.makeCustomHeaderToNet();
            RemotingCommand response = ackMessageProcessor.processRequest(handlerContext, request);

            assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
        }
    }