@Override
    public void submitPopConsumeRequest(
        final List<MessageExt> msgs,
        final PopProcessQueue processQueue,
        final MessageQueue messageQueue) {
        final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
        if (msgs.size() <= consumeBatchSize) {
            ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
            try {
                this.consumeExecutor.submit(consumeRequest);
            } catch (RejectedExecutionException e) {
                this.submitConsumeRequestLater(consumeRequest);
            }
        } else {
            for (int total = 0; total < msgs.size(); ) {
                List<MessageExt> msgThis = new ArrayList<>(consumeBatchSize);
                for (int i = 0; i < consumeBatchSize; i++, total++) {
                    if (total < msgs.size()) {
                        msgThis.add(msgs.get(total));
                    } else {
                        break;
                    }
                }

                ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
                try {
                    this.consumeExecutor.submit(consumeRequest);
                } catch (RejectedExecutionException e) {
                    for (; total < msgs.size(); total++) {
                        msgThis.add(msgs.get(total));
                    }

                    this.submitConsumeRequestLater(consumeRequest);
                }
            }
        }
    }
@Test
    public void testSubmitPopConsumeRequestWithMultiMsg() throws IllegalAccessException {
        List<MessageExt> msgs = Arrays.asList(createMessageExt(), createMessageExt());
        PopProcessQueue processQueue = mock(PopProcessQueue.class);
        MessageQueue messageQueue = mock(MessageQueue.class);
        ThreadPoolExecutor consumeExecutor = mock(ThreadPoolExecutor.class);
        FieldUtils.writeDeclaredField(popService, "consumeExecutor", consumeExecutor, true);
        when(defaultMQPushConsumer.getConsumeMessageBatchMaxSize()).thenReturn(1);
        popService.submitPopConsumeRequest(msgs, processQueue, messageQueue);
        verify(consumeExecutor, times(2)).submit(any(Runnable.class));
    }