@Override
    public long skip(long toSkip) throws IOException {
        getBufIfOpen(); // Check for closed stream
        if (toSkip <= 0) {
            return 0;
        }

        long remaining = toSkip;

        // Skip bytes stored in intermediate buffer first
        int avail = count - pos;
        int bytesSkipped = (int) Math.min(avail, remaining);
        pos += bytesSkipped;
        remaining -= bytesSkipped;

        while (remaining > 0) {
            if (delegateSkipToSourceStream) {
                // Use sourceStream's skip() to skip the rest.
                // conversion to int is acceptable because toSkip and remaining are int.
                long delegateBytesSkipped = getInIfOpen().skip(remaining);
                if (delegateBytesSkipped == 0) {
                    // read one byte to check for EOS
                    if (read() == -1) {
                        break;
                    }
                    // one byte read so decrement number to skip
                    remaining--;
                } else if (delegateBytesSkipped > remaining || delegateBytesSkipped < 0) { // skipped negative or too many bytes
                    throw new IOException("Unable to skip exactly");

                }
                remaining -= delegateBytesSkipped;
            } else {
                // skip from intermediate buffer, filling it first (if required)
                if (pos >= count) {
                    fill();
                    // if we don't have data in intermediate buffer after fill, then stop skipping
                    if (pos >= count)
                        break;
                }
                avail = count - pos;
                bytesSkipped = (int) Math.min(avail, remaining);
                pos += bytesSkipped;
                remaining -= bytesSkipped;
            }
        }
        return toSkip - remaining;
    }
Skips over and discards exactly {@code n} bytes of data from this input stream.
If {@code n} is zero, then no bytes are skipped.
If {@code n} is negative, then no bytes are skipped.
<p>
This method blocks until the requested number of bytes has been skipped, end of file is reached, or an
exception is thrown.
<p> If end of stream is reached before the stream is at the desired position, then the bytes skipped till than pointed
are returned.
<p> If an I/O error occurs, then the input stream may be in an inconsistent state. It is strongly recommended that the
stream be promptly closed if an I/O error occurs.
<p>
This method first skips and discards bytes in the intermediate buffer.
After that, depending on the value of {@link #delegateSkipToSourceStream}, it either delegates the skipping of
bytes to the sourceStream or it reads the data from input stream in chunks, copies the data into intermediate
buffer and skips it.
<p>
Starting JDK 12, a new method was introduced in InputStream, skipNBytes which has a similar behaviour as
this method.

@param toSkip the number of bytes to be skipped.
@return the actual number of bytes skipped which might be zero.
@throws IOException if this input stream has been closed by invoking its {@link #close()} method,
                    {@code in.skip(n)} throws an IOException, or an I/O error occurs.
clients/src/main/java/org/apache/kafka/common/utils/ChunkedBytesStream.java