public static DefaultRecord readFrom(InputStream input,
                                         long baseOffset,
                                         long baseTimestamp,
                                         int baseSequence,
                                         Long logAppendTime) throws IOException {
        int sizeOfBodyInBytes = ByteUtils.readVarint(input);
        ByteBuffer recordBuffer = ByteBuffer.allocate(sizeOfBodyInBytes);
        int bytesRead = Utils.readFully(input, recordBuffer);
        if (bytesRead != sizeOfBodyInBytes)
            throw new InvalidRecordException("Invalid record size: expected " + sizeOfBodyInBytes +
                " bytes in record payload, but the record payload reached EOF.");
        recordBuffer.flip(); // prepare for reading
        return readFrom(recordBuffer, sizeOfBodyInBytes, baseOffset, baseTimestamp,
                baseSequence, logAppendTime);
    }
Write the record to `out` and return its size.
clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java