《Java 消息中间件·Kafka 内核与客户端》8.2 Fetcher 与拉取管线

private ConsumerRecords<K, V> poll(final Timer timer) {
    acquireAndEnsureOpen();
    try {
        this.kafkaConsumerMetrics.recordPollStart(timer.currentTimeMs());

        if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) {
            throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
        }

        do {
            client.maybeTriggerWakeup();

            // try to update assignment metadata BUT do not need to block on the timer for join group
            updateAssignmentMetadataIfNeeded(timer, false);

            final Fetch<K, V> fetch = pollForFetches(timer);
            if (!fetch.isEmpty()) {
                // before returning the fetched records, we can send off the next round of fetches
                // and avoid block waiting for their responses to enable pipelining while the user
                // is handling the fetched records.
                //
                // NOTE: since the consumed position has already been updated, we must not allow
                // wakeups or any other errors to be triggered prior to returning the fetched records.
                if (sendFetches() > 0 || client.hasPendingRequests()) {
                    client.transmitSends();
                }

                if (fetch.records().isEmpty()) {
                    log.trace("Returning empty records from `poll()` "
                            + "since the consumer's position has advanced for at least one topic partition");
                }

                return this.interceptors.onConsume(new ConsumerRecords<>(fetch.records(), fetch.nextOffsets()));
            }
        } while (timer.notExpired());

        return ConsumerRecords.empty();
    } finally {
        release();
        this.kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
    }
}

private Fetch<K, V> pollForFetches(Timer timer) {
    long pollTimeout = coordinator == null ? timer.remainingMs() :
            Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs());

    // if data is available already, return it immediately
    final Fetch<K, V> fetch = fetcher.collectFetch();
    if (!fetch.isEmpty()) {
        return fetch;
    }

    // send any new fetches (won't resend pending fetches)
    sendFetches();

    // We do not want to be stuck blocking in poll if we are missing some positions
    // since the offset lookup may be backing off after a failure

    // NOTE: the use of cachedSubscriptionHasAllFetchPositions means we MUST call
    // updateAssignmentMetadataIfNeeded before this method.
    if (!cachedSubscriptionHasAllFetchPositions && pollTimeout > retryBackoffMs) {
        pollTimeout = retryBackoffMs;
    }

    log.trace("Polling for fetches with timeout {}", pollTimeout);

    timer.update(pollTimer.currentTimeMs());

    return fetcher.collectFetch();
}

collectFetch()  -> 本地有数据就立刻返回
sendFetches()   -> 没数据才发请求
client.poll()   -> 阻塞等响应，直到有可用数据或超时
collectFetch()  -> 再取一次

protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
    // Update metrics in case there was an assignment change
    metricsManager.maybeUpdateAssignment(subscriptions);

    Map<Node, FetchSessionHandler.Builder> fetchable = new HashMap<>();
    long currentTimeMs = time.milliseconds();
    Map<String, Uuid> topicIds = metadata.topicIds();

    // This is the set of partitions that have buffered data
    Set<TopicPartition> buffered = Collections.unmodifiableSet(fetchBuffer.bufferedPartitions());

    // This is the list of partitions that are fetchable and have no buffered data
    List<TopicPartition> unbuffered = fetchablePartitions(buffered);

    if (unbuffered.isEmpty()) {
        // If there are no partitions that don't already have data locally buffered, there's no need to issue
        // any fetch requests at the present time.
        return Collections.emptyMap();
    }

    Set<Integer> bufferedNodes = bufferedNodes(buffered, currentTimeMs);

    for (TopicPartition partition : unbuffered) {
        SubscriptionState.FetchPosition position = positionForPartition(partition);
        Optional<Node> nodeOpt = maybeNodeForPosition(partition, position, currentTimeMs);

        if (nodeOpt.isEmpty())
            continue;

        Node node = nodeOpt.get();

        } else if (bufferedNodes.contains(node.id())) {
            // While a node has buffered data, don't fetch other partition data from it. Because the buffered
            // partitions are not included in the fetch request, those partitions will be inadvertently dropped
            // from the broker fetch session cache. In some cases, that could lead to the entire fetch session
            // being evicted.
            log.trace("Skipping fetch for partition {} because its leader node {} hosts buffered partitions", partition, node);

        } else {
            // if there is a leader and no in-flight requests, issue a new fetch
            FetchSessionHandler.Builder builder = fetchable.computeIfAbsent(node, k -> {
                FetchSessionHandler fetchSessionHandler = sessionHandlers.computeIfAbsent(node.id(), n -> new FetchSessionHandler(logContext, n));
                return fetchSessionHandler.newBuilder();
            });
            Uuid topicId = topicIds.getOrDefault(partition.topic(), Uuid.ZERO_UUID);
            FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId,
                    position.offset,
                    FetchRequest.INVALID_LOG_START_OFFSET,
                    fetchConfig.fetchSize,
                    position.currentLeader.epoch,
                    Optional.empty());
            builder.add(partition, partitionData);

public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPartition> isAvailable) {
    // Since this is in the hot-path for fetching, we do this instead of using java.util.stream API
    List<TopicPartition> result = new ArrayList<>();
    assignment.forEach((topicPartition, topicPartitionState) -> {
        // Cheap check is first to avoid evaluating the predicate if possible
        if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
                && isAvailable.test(topicPartition)) {
            result.add(topicPartition);
        }
    });
    return result;
}

// we move the partition to the end if we received some bytes. This way, it's more likely that partitions
// for the same topic can remain together (allowing for more efficient serialization).
if (bytesRead > 0)
    subscriptions.movePartitionToEnd(partition);

private static final String MAX_POLL_RECORDS_DOC = "The maximum number of records returned in a single call to poll()."
    + " Note, that <code>" + MAX_POLL_RECORDS_CONFIG + "</code> does not impact the underlying fetching behavior."
    + " The consumer will cache the records from each fetch request and returns them incrementally from each poll.";
public static final int DEFAULT_MAX_POLL_RECORDS = 500;

public Fetch<K, V> collectFetch(final FetchBuffer fetchBuffer) {
    final Fetch<K, V> fetch = Fetch.empty();
    final Queue<CompletedFetch> pausedCompletedFetches = new ArrayDeque<>();
    int recordsRemaining = fetchConfig.maxPollRecords;

    try {
        while (recordsRemaining > 0) {
            final CompletedFetch nextInLineFetch = fetchBuffer.nextInLineFetch();

            } else {
                final Fetch<K, V> nextFetch = fetchRecords(nextInLineFetch, recordsRemaining);
                recordsRemaining -= nextFetch.numRecords();
                fetch.add(nextFetch);
            }
        }

poll #01 -> count=0累计=0 position={}
（#02–#06 均为 count=0，加入组期间）
poll #07 -> count=5累计=5 position={b3.lab.fetch-0=5}
poll #08 -> count=5累计=10 position={b3.lab.fetch-0=10}
...
poll #25 -> count=5累计=95 position={b3.lab.fetch-0=95}
poll #26 -> count=5累计=100 position={b3.lab.fetch-0=100}
poll #27 -> count=0累计=100 position={b3.lab.fetch-0=100}
total=100

final FetchRequest.Builder request = FetchRequest.Builder
        .forConsumer(maxVersion, fetchConfig.maxWaitMs, fetchConfig.minBytes, requestData.toSend())
        .isolationLevel(fetchConfig.isolationLevel)
        .setMaxBytes(fetchConfig.maxBytes)

poll #01 count=0 payloadBytes=0 累计=0
（#02–#06 同为 0，加入组期间）
poll #07 count=16 payloadBytes=16000 累计=16
（#08–#11 每次 16 条、16000 字节，略）
poll #12 count=16 payloadBytes=16000 累计=96
total=96