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

Code before:
                    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);

                log.debug("Added {} fetch request for partition {} at position {} to node {}", fetchConfig.isolationLevel,
                        partition, position, node);
            }
        }

        return convert(fetchable);
    }

    private Map<Node, FetchSessionHandler.FetchRequestData> convert(Map<Node, FetchSessionHandler.Builder> fetchable) {
        Map<Node, FetchSessionHandler.FetchRequestData> map = new HashMap<>(fetchable.size());

        for (Map.Entry<Node, FetchSessionHandler.Builder> entry : fetchable.entrySet()) {
            Node node = entry.getKey();
            FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();
            map.put(node, fetchRequestData);
        }

        return map;
    }

    /**
     * Simple utility method that returns a {@link SubscriptionState.FetchPosition position} for the partition. If
     * no position exists, an {@link IllegalStateException} is thrown.
     */
    private SubscriptionState.FetchPosition positionForPartition(TopicPartition partition) {
        SubscriptionState.FetchPosition position = subscriptions.position(partition);

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

        return position;
    }

    /**
     * Retrieves the node from which to fetch the partition data. If the given
     * {@link SubscriptionState.FetchPosition position} does not have a current
     * {@link Metadata.LeaderAndEpoch#leader leader} defined the method will return {@link Optional#empty()}.
     *
     * @return Three options: 1) {@link Optional#empty()} if the position's leader is empty, 2) the
     * {@link #selectReadReplica(TopicPartition, Node, long) read replica, if defined}, or 3) the position's

Review comment (suggestion):
```suggestion
            map.put(entry.getKey(), entry.getValue().build());
```

Code after:
                    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);

                log.debug("Added {} fetch request for partition {} at position {} to node {}", fetchConfig.isolationLevel,
                        partition, position, node);
            }
        }

        return convert(fetchable);
    }

    /**
     * This method converts {@link FetchSessionHandler.Builder} instances to
     * {@link FetchSessionHandler.FetchRequestData} instances. It intentionally forgoes use of the Java Collections
     * Streams API to reduce overhead in the critical network path.
     */
    private Map<Node, FetchSessionHandler.FetchRequestData> convert(Map<Node, FetchSessionHandler.Builder> fetchable) {
        Map<Node, FetchSessionHandler.FetchRequestData> map = new HashMap<>(fetchable.size());

        for (Map.Entry<Node, FetchSessionHandler.Builder> entry : fetchable.entrySet())
            map.put(entry.getKey(), entry.getValue().build());

        return map;
    }

    /**
     * Simple utility method that returns a {@link SubscriptionState.FetchPosition position} for the partition. If
     * no position exists, an {@link IllegalStateException} is thrown.
     */
    private SubscriptionState.FetchPosition positionForPartition(TopicPartition partition) {
        SubscriptionState.FetchPosition position = subscriptions.position(partition);

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

        return position;
    }

    /**
     * Retrieves the node from which to fetch the partition data. If the given
     * {@link SubscriptionState.FetchPosition position} does not have a current
     * {@link Metadata.LeaderAndEpoch#leader leader} defined the method will return {@link Optional#empty()}.
     *