Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java (Java)

Code before:
                log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
                        fetchConfig.isolationLevel, fetchOffset, partition, partitionData);

                Errors partitionError = Errors.forCode(partitionData.errorCode());
                if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
                    log.debug("For {}, received error {}, with leaderIdAndEpoch {}", partition, partitionError, partitionData.currentLeader());
                    if (partitionData.currentLeader().leaderId() != -1 && partitionData.currentLeader().leaderEpoch() != -1) {
                        partitionsWithUpdatedLeaderInfo.put(partition, new Metadata.LeaderIdAndEpoch(
                            Optional.of(partitionData.currentLeader().leaderId()), Optional.of(partitionData.currentLeader().leaderEpoch())));
                    }
                }

                CompletedFetch completedFetch = new CompletedFetch(
                        completedFetchLog,
                        subscriptions,
                        decompressionBufferSupplier,
                        partition,
                        partitionData,
                        metricAggregator,
                        fetchOffset);
                fetchBuffer.add(completedFetch);
            }

            // "Wake" the fetch buffer on any response, even if it's empty, to allow the consumer to not block
            // indefinitely waiting on the fetch buffer to get data.
            fetchBuffer.wakeup();

            if (!partitionsWithUpdatedLeaderInfo.isEmpty()) {
                List<Node> leaderNodes = new ArrayList<>();

                for (FetchResponseData.NodeEndpoint e : response.data().nodeEndpoints()) {
                    Node node = new Node(e.nodeId(), e.host(), e.port(), e.rack());

                    if (!node.equals(Node.noNode()))
                        leaderNodes.add(node);
                }

                Set<TopicPartition> updatedPartitions = metadata.updatePartitionLeadership(partitionsWithUpdatedLeaderInfo, leaderNodes);
                updatedPartitions.forEach(
                    tp -> {
                        log.debug("For {}, as the leader was updated, position will be validated.", tp);
                        subscriptions.maybeValidatePositionForCurrentLeader(apiVersions, tp, metadata.currentLeader(tp));
                    }
                );
            }

            metricsManager.recordLatency(resp.destination(), resp.requestLatencyMs());
        } finally {
            removePendingFetchRequest(fetchTarget, data.metadata().sessionId());
        }
    }

Review comment (performance):
Should we notify the buffer as soon as we add the first completed batch to it, instead of waiting for all completed batches to be added? 

Seems that will be closer to what the `ClassicConsumer` does by polling while `!fetcher.hasAvailableFetches()` (given that hasAvailable fetches will be true as soon as there is a first CompletedBatch added)

Code after:
                long fetchOffset = requestData.fetchOffset;
                FetchResponseData.PartitionData partitionData = entry.getValue();

                log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
                        fetchConfig.isolationLevel, fetchOffset, partition, partitionData);

                Errors partitionError = Errors.forCode(partitionData.errorCode());
                if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
                    log.debug("For {}, received error {}, with leaderIdAndEpoch {}", partition, partitionError, partitionData.currentLeader());
                    if (partitionData.currentLeader().leaderId() != -1 && partitionData.currentLeader().leaderEpoch() != -1) {
                        partitionsWithUpdatedLeaderInfo.put(partition, new Metadata.LeaderIdAndEpoch(
                            Optional.of(partitionData.currentLeader().leaderId()), Optional.of(partitionData.currentLeader().leaderEpoch())));
                    }
                }

                CompletedFetch completedFetch = new CompletedFetch(
                        completedFetchLog,
                        subscriptions,
                        decompressionBufferSupplier,
                        partition,
                        partitionData,
                        metricAggregator,
                        fetchOffset);
                fetchBuffer.add(completedFetch);
                needsWakeup = false;
            }

            // "Wake" the fetch buffer on any response, even if it's empty, to allow the consumer to not block
            // indefinitely waiting on the fetch buffer to get data.
            if (needsWakeup)
                fetchBuffer.wakeup();

            if (!partitionsWithUpdatedLeaderInfo.isEmpty()) {
                List<Node> leaderNodes = new ArrayList<>();

                for (FetchResponseData.NodeEndpoint e : response.data().nodeEndpoints()) {
                    Node node = new Node(e.nodeId(), e.host(), e.port(), e.rack());

                    if (!node.equals(Node.noNode()))
                        leaderNodes.add(node);
                }

                Set<TopicPartition> updatedPartitions = metadata.updatePartitionLeadership(partitionsWithUpdatedLeaderInfo, leaderNodes);
                updatedPartitions.forEach(
                    tp -> {
                        log.debug("For {}, as the leader was updated, position will be validated.", tp);
                        subscriptions.maybeValidatePositionForCurrentLeader(apiVersions, tp, metadata.currentLeader(tp));
                    }
                );
            }