public void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest) {

        if (consumeRequest.getMsgs().isEmpty()) {
            return;
        }

        int ackIndex = context.getAckIndex();
        String topic = consumeRequest.getMessageQueue().getTopic();

        switch (status) {
            case CONSUME_SUCCESS:
                if (ackIndex >= consumeRequest.getMsgs().size()) {
                    ackIndex = consumeRequest.getMsgs().size() - 1;
                }
                int ok = ackIndex + 1;
                int failed = consumeRequest.getMsgs().size() - ok;
                this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, topic, ok);
                this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, topic, failed);
                break;
            case RECONSUME_LATER:
                ackIndex = -1;
                this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, topic,
                        consumeRequest.getMsgs().size());
                break;
            default:
                break;
        }

        //ack if consume success
        for (int i = 0; i <= ackIndex; i++) {
            this.defaultMQPushConsumerImpl.ackAsync(consumeRequest.getMsgs().get(i), consumerGroup);
            consumeRequest.getPopProcessQueue().ack();
        }

        //consume later if consume fail
        for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
            MessageExt msgExt = consumeRequest.getMsgs().get(i);
            consumeRequest.getPopProcessQueue().ack();
            if (msgExt.getReconsumeTimes() >= this.defaultMQPushConsumerImpl.getMaxReconsumeTimes()) {
                checkNeedAckOrDelay(msgExt);
                continue;
            }

            int delayLevel = context.getDelayLevelWhenNextConsume();
            changePopInvisibleTime(consumeRequest.getMsgs().get(i), consumerGroup, delayLevel);
        }
    }
@Test
    public void testProcessConsumeResult() {
        ConsumeConcurrentlyContext context = mock(ConsumeConcurrentlyContext.class);
        ConsumeMessagePopConcurrentlyService.ConsumeRequest consumeRequest = mock(ConsumeMessagePopConcurrentlyService.ConsumeRequest.class);
        when(consumeRequest.getMsgs()).thenReturn(Arrays.asList(createMessageExt(), createMessageExt()));
        MessageQueue messageQueue = mock(MessageQueue.class);
        when(messageQueue.getTopic()).thenReturn(defaultTopic);
        when(consumeRequest.getMessageQueue()).thenReturn(messageQueue);
        PopProcessQueue processQueue = mock(PopProcessQueue.class);
        when(processQueue.ack()).thenReturn(0);
        when(consumeRequest.getPopProcessQueue()).thenReturn(processQueue);
        when(defaultMQPushConsumerImpl.getPopDelayLevel()).thenReturn(new int[]{1, 10});
        popService.processConsumeResult(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, context, consumeRequest);
        verify(defaultMQPushConsumerImpl, times(1)).ackAsync(any(MessageExt.class), any());
    }