dajac commented on a change in pull request #9008:
URL: https://github.com/apache/kafka/pull/9008#discussion_r454866708
##########
File path:
clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
##########
@@ -273,6 +99,28 @@ public boolean equals(Object o) {
}
}
+ private Map<TopicPartition, PartitionData>
toPartitionDataMap(List<FetchRequestData.FetchTopic> fetchableTopics) {
+ Map<TopicPartition, PartitionData> result = new LinkedHashMap<>();
+ fetchableTopics.forEach(fetchTopic ->
fetchTopic.partitions().forEach(fetchPartition -> {
+ Optional<Integer> leaderEpoch =
Optional.of(fetchPartition.currentLeaderEpoch())
+ .filter(epoch -> epoch !=
RecordBatch.NO_PARTITION_LEADER_EPOCH);
+ result.put(new TopicPartition(fetchTopic.topic(),
fetchPartition.partition()),
+ new PartitionData(fetchPartition.fetchOffset(),
fetchPartition.logStartOffset(),
+ fetchPartition.partitionMaxBytes(), leaderEpoch));
Review comment:
@mumrah Have we considered dropping the `PartitionData` class entirely
in favour of using `FetchRequestData .FetchPartition` directly in the broker?
The main difference is that `FetchPartition` does not have an `Optional` for
the leader epoch but returns the default value (-1) instead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
