private Record nextFetchedRecord(final boolean checkCrcs) {
        while (true) {
            if (records == null || !records.hasNext()) {
                maybeCloseRecordStream();

                if (!batches.hasNext()) {
                    drain();
                    return null;
                }

                currentBatch = batches.next();
                maybeEnsureValid(currentBatch, checkCrcs);

                records = currentBatch.streamingIterator(decompressionBufferSupplier);
            } else {
                Record record = records.next();
                maybeEnsureValid(record, checkCrcs);

                // control records are not returned to the user
                if (!currentBatch.isControlBatch()) {
                    return record;
                }
            }
        }
    }
Parse the record entry, deserializing the key / value fields if necessary
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java