private static int sizeOf(int keySize, int valueSize, Header[] headers) {
        int size = 0;
        if (keySize < 0)
            size += NULL_VARINT_SIZE_BYTES;
        else
            size += ByteUtils.sizeOfVarint(keySize) + keySize;

        if (valueSize < 0)
            size += NULL_VARINT_SIZE_BYTES;
        else
            size += ByteUtils.sizeOfVarint(valueSize) + valueSize;

        if (headers == null)
            throw new IllegalArgumentException("Headers cannot be null");

        size += ByteUtils.sizeOfVarint(headers.length);
        for (Header header : headers) {
            String headerKey = header.key();
            if (headerKey == null)
                throw new IllegalArgumentException("Invalid null header key found in headers");

            int headerKeySize = Utils.utf8Length(headerKey);
            size += ByteUtils.sizeOfVarint(headerKeySize) + headerKeySize;

            byte[] headerValue = header.value();
            if (headerValue == null) {
                size += NULL_VARINT_SIZE_BYTES;
            } else {
                size += ByteUtils.sizeOfVarint(headerValue.length) + headerValue.length;
            }
        }
        return size;
    }
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