public synchronized void rewind() {
        this.readPosition = 0;
        this.curReadBufferIndex = 0;
        this.readPosInCurBuffer = 0;
        if (CollectionUtils.isNotEmpty(bufferList)) {
            this.curBuffer = bufferList.get(0);
            for (ByteBuffer buffer : bufferList) {
                buffer.rewind();
            }
        }
    }
@Test
    public void testConsumeQueueTypeInputStream() {
        List<ByteBuffer> uploadBufferList = new ArrayList<>();
        int bufferSize = 0;
        for (int i = 0; i < MSG_NUM; i++) {
            ByteBuffer byteBuffer = MessageFormatUtilTest.buildMockedConsumeQueueBuffer();
            uploadBufferList.add(byteBuffer);
            bufferSize += byteBuffer.remaining();
        }

        // build expected byte buffer for verifying the FileSegmentInputStream
        ByteBuffer expectedByteBuffer = ByteBuffer.allocate(bufferSize);
        for (ByteBuffer byteBuffer : uploadBufferList) {
            expectedByteBuffer.put(byteBuffer);
            byteBuffer.rewind();
        }

        int finalBufferSize = bufferSize;
        int[] batchReadSizeTestSet = {MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE - 1, MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE, MessageFormatUtil.CONSUME_QUEUE_UNIT_SIZE + 1};
        verifyReadAndReset(expectedByteBuffer, () -> FileSegmentInputStreamFactory.build(
            FileSegmentType.CONSUME_QUEUE, COMMIT_LOG_START_OFFSET, uploadBufferList, null, finalBufferSize), bufferSize, batchReadSizeTestSet);
    }