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

            BatchAck bAck1 = new BatchAck();
            bAck1.setConsumerGroup(MixAll.DEFAULT_CONSUMER_GROUP);
            bAck1.setTopic(topic);
            bAck1.setStartOffset(MIN_OFFSET_IN_QUEUE);
            bAck1.setBitSet(new BitSet());
            bAck1.getBitSet().set(1);
            bAck1.setRetry("0");

            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.BATCH_ACK_MESSAGE, null);
            BatchAckMessageRequestBody reqBody = new BatchAckMessageRequestBody();
            reqBody.setAcks(Collections.singletonList(bAck1));
            request.setBody(reqBody.encode());
            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);

            BatchAck bAck1 = new BatchAck();
            bAck1.setConsumerGroup(MixAll.DEFAULT_CONSUMER_GROUP);
            bAck1.setTopic(topic);
            bAck1.setStartOffset(MIN_OFFSET_IN_QUEUE);
            bAck1.setBitSet(new BitSet());
            bAck1.getBitSet().set(1);
            bAck1.setRetry("0");

            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.BATCH_ACK_MESSAGE, null);
            BatchAckMessageRequestBody reqBody = new BatchAckMessageRequestBody();
            reqBody.setAcks(Arrays.asList(bAck1));
            request.setBody(reqBody.encode());
            request.makeCustomHeaderToNet();
            RemotingCommand response = ackMessageProcessor.processRequest(handlerContext, request);

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