private BatchAndType nextControlBatch(FileChannelRecordBatch input) {
        List<ApiMessageAndVersion> messages = new ArrayList<>();
        for (Record record : input) {
            try {
                short typeId = ControlRecordType.parseTypeId(record.key());
                ControlRecordType type = ControlRecordType.fromTypeId(typeId);
                switch (type) {
                    case LEADER_CHANGE: {
                        LeaderChangeMessage message = new LeaderChangeMessage();
                        message.read(new ByteBufferAccessor(record.value()), (short) 0);
                        messages.add(new ApiMessageAndVersion(message, (short) 0));
                        break;
                    }
                    case SNAPSHOT_HEADER: {
                        SnapshotHeaderRecord message = new SnapshotHeaderRecord();
                        message.read(new ByteBufferAccessor(record.value()), (short) 0);
                        messages.add(new ApiMessageAndVersion(message, (short) 0));
                        break;
                    }
                    case SNAPSHOT_FOOTER: {
                        SnapshotFooterRecord message = new SnapshotFooterRecord();
                        message.read(new ByteBufferAccessor(record.value()), (short) 0);
                        messages.add(new ApiMessageAndVersion(message, (short) 0));
                        break;
                    }
                    default:
                        throw new RuntimeException("Unsupported control record type " + type + " at offset " +
                                record.offset());
                }
            } catch (Throwable e) {
                throw new RuntimeException("Unable to read control record at offset " + record.offset(), e);
            }
        }
        return new BatchAndType(Batch.data(
            input.baseOffset(),
            input.partitionLeaderEpoch(),
            input.maxTimestamp(),
            input.sizeInBytes(),
            messages), true);
    }
Reads a log file containing KRaft record batches.
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java