public MutableRecordBatch nextBatch() {
        int remaining = buffer.remaining();

        Integer batchSize = nextBatchSize();
        if (batchSize == null || remaining < batchSize)
            return null;

        byte magic = buffer.get(buffer.position() + MAGIC_OFFSET);

        ByteBuffer batchSlice = buffer.slice();
        batchSlice.limit(batchSize);
        buffer.position(buffer.position() + batchSize);

        if (magic > RecordBatch.MAGIC_VALUE_V1)
            return new DefaultRecordBatch(batchSlice);
        else
            return new AbstractLegacyRecordBatch.ByteBufferLegacyRecordBatch(batchSlice);
    }
@Test
    public void iteratorRaisesOnTooSmallRecords() {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, Compression.NONE, TimestampType.CREATE_TIME, 0L);
        builder.append(15L, "a".getBytes(), "1".getBytes());
        builder.append(20L, "b".getBytes(), "2".getBytes());
        builder.close();

        int position = buffer.position();

        builder = MemoryRecords.builder(buffer, Compression.NONE, TimestampType.CREATE_TIME, 2L);
        builder.append(30L, "c".getBytes(), "3".getBytes());
        builder.append(40L, "d".getBytes(), "4".getBytes());
        builder.close();

        buffer.flip();
        buffer.putInt(position + DefaultRecordBatch.LENGTH_OFFSET, 9);

        ByteBufferLogInputStream logInputStream = new ByteBufferLogInputStream(buffer, Integer.MAX_VALUE);
        assertNotNull(logInputStream.nextBatch());
        assertThrows(CorruptRecordException.class, logInputStream::nextBatch);
    }