private BatchAndType nextDataBatch(FileChannelRecordBatch input) {
        List<ApiMessageAndVersion> messages = new ArrayList<>();
        for (Record record : input) {
            try {
                ByteBufferAccessor accessor = new ByteBufferAccessor(record.value());
                ApiMessageAndVersion messageAndVersion = serde.read(accessor, record.valueSize());
                messages.add(messageAndVersion);
            } catch (Throwable e) {
                throw new RuntimeException("unable to deserialize record at offset " + record.offset(), e);
            }
        }
        return new BatchAndType(Batch.data(
            input.baseOffset(),
            input.partitionLeaderEpoch(),
            input.maxTimestamp(),
            input.sizeInBytes(),
            messages), false);
    }
Reads a log file containing KRaft record batches.
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java