private Fetch<K, V> fetchRecords(final CompletedFetch nextInLineFetch, int maxRecords) {
        final TopicPartition tp = nextInLineFetch.partition;

        if (!subscriptions.isAssigned(tp)) {
            // this can happen when a rebalance happened before fetched records are returned to the consumer's poll call
            log.debug("Not returning fetched records for partition {} since it is no longer assigned", tp);
        } else if (!subscriptions.isFetchable(tp)) {
            // this can happen when a partition is paused before fetched records are returned to the consumer's
            // poll call or if the offset is being reset
            log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", tp);
        } else {
            SubscriptionState.FetchPosition position = subscriptions.position(tp);

            if (position == null)
                throw new IllegalStateException("Missing position for fetchable partition " + tp);

            if (nextInLineFetch.nextFetchOffset() == position.offset) {
                List<ConsumerRecord<K, V>> partRecords = nextInLineFetch.fetchRecords(fetchConfig,
                        deserializers,
                        maxRecords);

                log.trace("Returning {} fetched records at offset {} for assigned partition {}",
                        partRecords.size(), position, tp);

                boolean positionAdvanced = false;

                if (nextInLineFetch.nextFetchOffset() > position.offset) {
                    SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
                            nextInLineFetch.nextFetchOffset(),
                            nextInLineFetch.lastEpoch(),
                            position.currentLeader);
                    log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`",
                            position, nextPosition, tp, partRecords.size());
                    subscriptions.position(tp, nextPosition);
                    positionAdvanced = true;
                }

                Long partitionLag = subscriptions.partitionLag(tp, fetchConfig.isolationLevel);
                if (partitionLag != null)
                    metricsManager.recordPartitionLag(tp, partitionLag);

                Long lead = subscriptions.partitionLead(tp);
                if (lead != null) {
                    metricsManager.recordPartitionLead(tp, lead);
                }

                return Fetch.forPartition(tp, partRecords, positionAdvanced);
            } else {
                // these records aren't next in line based on the last consumed position, ignore them
                // they must be from an obsolete request
                log.debug("Ignoring fetched records for {} at offset {} since the current position is {}",
                        tp, nextInLineFetch.nextFetchOffset(), position);
            }
        }

        log.trace("Draining fetched records for partition {}", tp);
        nextInLineFetch.drain();

        return Fetch.empty();
    }
Return the fetched {@link ConsumerRecord records}, empty the {@link FetchBuffer record buffer}, and
update the consumed position.

</p>

NOTE: returning an {@link Fetch#empty() empty} fetch guarantees the consumed position is not updated.

@param fetchBuffer {@link FetchBuffer} from which to retrieve the {@link ConsumerRecord records}

@return A {@link Fetch} for the requested partitions
@throws OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse and
        the defaultResetPolicy is NONE
@throws TopicAuthorizationException If there is TopicAuthorization error in fetchResponse.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java