public static List<SelectBufferResult> splitMessageBuffer(ByteBuffer cqBuffer, ByteBuffer msgBuffer) {

        if (cqBuffer == null || msgBuffer == null) {
            log.error("MessageFormatUtil split buffer error, cq buffer or msg buffer is null");
            return new ArrayList<>();
        }

        cqBuffer.rewind();
        msgBuffer.rewind();

        List<SelectBufferResult> bufferResultList = new ArrayList<>(
            cqBuffer.remaining() / CONSUME_QUEUE_UNIT_SIZE);

        if (msgBuffer.remaining() == 0) {
            log.error("MessageFormatUtil split buffer error, msg buffer length is 0");
            return bufferResultList;
        }

        if (cqBuffer.remaining() == 0 || cqBuffer.remaining() % CONSUME_QUEUE_UNIT_SIZE != 0) {
            log.error("MessageFormatUtil split buffer error, cq buffer size is {}", cqBuffer.remaining());
            return bufferResultList;
        }

        try {
            long firstCommitLogOffset = MessageFormatUtil.getCommitLogOffsetFromItem(cqBuffer);

            for (int position = cqBuffer.position(); position < cqBuffer.limit();
                position += CONSUME_QUEUE_UNIT_SIZE) {

                cqBuffer.position(position);
                long logOffset = MessageFormatUtil.getCommitLogOffsetFromItem(cqBuffer);
                int bufferSize = MessageFormatUtil.getSizeFromItem(cqBuffer);
                long tagCode = MessageFormatUtil.getTagCodeFromItem(cqBuffer);

                int offset = (int) (logOffset - firstCommitLogOffset);
                if (offset + bufferSize > msgBuffer.limit()) {
                    log.error("MessageFormatUtil split buffer error, message buffer offset exceeded limit. " +
                        "Expect length: {}, Actual length: {}", offset + bufferSize, msgBuffer.limit());
                    break;
                }

                msgBuffer.position(offset);
                int magicCode = getMagicCode(msgBuffer);
                if (magicCode == BLANK_MAGIC_CODE) {
                    offset += COMMIT_LOG_CODA_SIZE;
                    msgBuffer.position(offset);
                    magicCode = getMagicCode(msgBuffer);
                }
                if (magicCode != MessageDecoder.MESSAGE_MAGIC_CODE &&
                    magicCode != MessageDecoder.MESSAGE_MAGIC_CODE_V2) {
                    log.error("MessageFormatUtil split buffer error, found unknown magic code. " +
                        "Message offset: {}, wrong magic code: {}", offset, magicCode);
                    continue;
                }

                if (bufferSize != getTotalSize(msgBuffer)) {
                    log.error("MessageFormatUtil split buffer error, message length not match. " +
                        "CommitLog length: {}, buffer length: {}", getTotalSize(msgBuffer), bufferSize);
                    continue;
                }

                ByteBuffer sliceBuffer = msgBuffer.slice();
                sliceBuffer.limit(bufferSize);
                bufferResultList.add(new SelectBufferResult(sliceBuffer, offset, bufferSize, tagCode));
            }
        } finally {
            cqBuffer.rewind();
            msgBuffer.rewind();
        }
        return bufferResultList;
    }
@Test
    public void testSplitMessages() {
        ByteBuffer msgBuffer1 = buildMockedMessageBuffer();
        msgBuffer1.putLong(MessageFormatUtil.QUEUE_OFFSET_POSITION, 10);

        ByteBuffer msgBuffer2 = ByteBuffer.allocate(COMMIT_LOG_CODA_SIZE);
        msgBuffer2.putInt(MessageFormatUtil.COMMIT_LOG_CODA_SIZE);
        msgBuffer2.putInt(MessageFormatUtil.BLANK_MAGIC_CODE);
        msgBuffer2.putLong(System.currentTimeMillis());
        msgBuffer2.flip();

        ByteBuffer msgBuffer3 = buildMockedMessageBuffer();
        msgBuffer3.putLong(MessageFormatUtil.QUEUE_OFFSET_POSITION, 11);

        ByteBuffer msgBuffer = ByteBuffer.allocate(
            msgBuffer1.remaining() + msgBuffer2.remaining() + msgBuffer3.remaining());
        msgBuffer.put(msgBuffer1);
        msgBuffer.put(msgBuffer2);
        msgBuffer.put(msgBuffer3);
        msgBuffer.flip();

        ByteBuffer cqBuffer1 = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer1.putLong(1000);
        cqBuffer1.putInt(MSG_LEN);
        cqBuffer1.putLong(0);
        cqBuffer1.flip();

        ByteBuffer cqBuffer2 = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer2.putLong(1000 + MessageFormatUtil.COMMIT_LOG_CODA_SIZE + MSG_LEN);
        cqBuffer2.putInt(MSG_LEN);
        cqBuffer2.putLong(0);
        cqBuffer2.flip();

        ByteBuffer cqBuffer3 = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer3.putLong(1000 + MSG_LEN);
        cqBuffer3.putInt(MSG_LEN);
        cqBuffer3.putLong(0);
        cqBuffer3.flip();

        ByteBuffer cqBuffer4 = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer4.putLong(1000 + MessageFormatUtil.COMMIT_LOG_CODA_SIZE + MSG_LEN);
        cqBuffer4.putInt(MSG_LEN - 10);
        cqBuffer4.putLong(0);
        cqBuffer4.flip();

        ByteBuffer cqBuffer5 = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer5.putLong(1000 + MessageFormatUtil.COMMIT_LOG_CODA_SIZE + MSG_LEN);
        cqBuffer5.putInt(MSG_LEN * 10);
        cqBuffer5.putLong(0);
        cqBuffer5.flip();

        // Message buffer size is 0 or consume queue buffer size is 0
        Assert.assertEquals(0,
            MessageFormatUtil.splitMessageBuffer(null, ByteBuffer.allocate(0)).size());
        Assert.assertEquals(0,
            MessageFormatUtil.splitMessageBuffer(cqBuffer1, null).size());
        Assert.assertEquals(0,
            MessageFormatUtil.splitMessageBuffer(cqBuffer1, ByteBuffer.allocate(0)).size());
        Assert.assertEquals(0,
            MessageFormatUtil.splitMessageBuffer(ByteBuffer.allocate(0), msgBuffer).size());
        Assert.assertEquals(0,
            MessageFormatUtil.splitMessageBuffer(ByteBuffer.allocate(10), msgBuffer).size());

        ByteBuffer cqBuffer = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE * 2);
        cqBuffer.put(cqBuffer1);
        cqBuffer.put(cqBuffer2);
        cqBuffer.flip();
        cqBuffer1.rewind();
        cqBuffer2.rewind();
        List<SelectBufferResult> msgList = MessageFormatUtil.splitMessageBuffer(cqBuffer, msgBuffer);
        Assert.assertEquals(2, msgList.size());
        Assert.assertEquals(0, msgList.get(0).getStartOffset());
        Assert.assertEquals(MSG_LEN, msgList.get(0).getSize());
        Assert.assertEquals(MSG_LEN + MessageFormatUtil.COMMIT_LOG_CODA_SIZE, msgList.get(1).getStartOffset());
        Assert.assertEquals(MSG_LEN, msgList.get(1).getSize());

        cqBuffer = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE * 2);
        cqBuffer.put(cqBuffer1);
        cqBuffer.put(cqBuffer4);
        cqBuffer.flip();
        cqBuffer1.rewind();
        cqBuffer4.rewind();
        msgList = MessageFormatUtil.splitMessageBuffer(cqBuffer, msgBuffer);
        Assert.assertEquals(1, msgList.size());
        Assert.assertEquals(0, msgList.get(0).getStartOffset());
        Assert.assertEquals(MSG_LEN, msgList.get(0).getSize());

        cqBuffer = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE * 3);
        cqBuffer.put(cqBuffer1);
        cqBuffer.put(cqBuffer3);
        cqBuffer.flip();
        cqBuffer1.rewind();
        cqBuffer3.rewind();
        msgList = MessageFormatUtil.splitMessageBuffer(cqBuffer, msgBuffer);
        Assert.assertEquals(2, msgList.size());
        Assert.assertEquals(0, msgList.get(0).getStartOffset());
        Assert.assertEquals(MSG_LEN, msgList.get(0).getSize());
        Assert.assertEquals(MSG_LEN + MessageFormatUtil.COMMIT_LOG_CODA_SIZE, msgList.get(1).getStartOffset());
        Assert.assertEquals(MSG_LEN, msgList.get(1).getSize());

        cqBuffer = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE);
        cqBuffer.put(cqBuffer5);
        cqBuffer.flip();
        msgList = MessageFormatUtil.splitMessageBuffer(cqBuffer, msgBuffer);
        Assert.assertEquals(0, msgList.size());

        // Wrong magic code, it will destroy the mocked message buffer
        msgBuffer.putInt(MessageFormatUtil.MAGIC_CODE_POSITION, -1);
        cqBuffer = ByteBuffer.allocate(MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE * 2);
        cqBuffer.put(cqBuffer1);
        cqBuffer.put(cqBuffer2);
        cqBuffer.flip();
        cqBuffer1.rewind();
        cqBuffer2.rewind();
        Assert.assertEquals(1, MessageFormatUtil.splitMessageBuffer(cqBuffer, msgBuffer).size());
    }