private static Header[] readHeaders(ByteBuffer buffer, int numHeaders) {
        Header[] headers = new Header[numHeaders];
        for (int i = 0; i < numHeaders; i++) {
            int headerKeySize = ByteUtils.readVarint(buffer);
            if (headerKeySize < 0)
                throw new InvalidRecordException("Invalid negative header key size " + headerKeySize);

            ByteBuffer headerKeyBuffer = Utils.readBytes(buffer, headerKeySize);

            int headerValueSize = ByteUtils.readVarint(buffer);
            ByteBuffer headerValue = Utils.readBytes(buffer, headerValueSize);

            headers[i] = new RecordHeader(headerKeyBuffer, headerValue);
        }

        return headers;
    }
Skips over and discards exactly {@code bytesToSkip} bytes from the input stream.

We require a loop over {@link InputStream#skip(long)} because it is possible for InputStream to skip smaller
number of bytes than expected (see javadoc for InputStream#skip).

No-op for case where bytesToSkip <= 0. This could occur for cases where field is expected to be null.
@throws InvalidRecordException if end of stream is encountered before we could skip required bytes.
@throws IOException is an I/O error occurs while trying to skip from InputStream.

@see java.io.InputStream#skip(long)
clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java