ijuma commented on a change in pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#discussion_r551407113
##########
File path:
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -331,28 +331,32 @@ public int sessionId() {
private static <T extends BaseRecords> FetchResponseData toMessage(int
throttleTimeMs, Errors error,
Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator,
int
sessionId) {
- FetchResponseData message = new FetchResponseData();
- message.setThrottleTimeMs(throttleTimeMs);
- message.setErrorCode(error.code());
- message.setSessionId(sessionId);
-
List<FetchResponseData.FetchableTopicResponse> topicResponseList = new
ArrayList<>();
- List<FetchRequest.TopicAndPartitionData<PartitionData<T>>> topicsData =
- FetchRequest.TopicAndPartitionData.batchByTopic(partIterator);
- topicsData.forEach(partitionDataTopicAndPartitionData -> {
- List<FetchResponseData.FetchablePartitionResponse>
partitionResponses = new ArrayList<>();
-
partitionDataTopicAndPartitionData.partitions.forEach((partitionId,
partitionData) -> {
- // Since PartitionData alone doesn't know the partition ID, we
set it here
- partitionData.partitionResponse.setPartition(partitionId);
+ partIterator.forEachRemaining(entry -> {
+ PartitionData<T> partitionData = entry.getValue();
+ // Since PartitionData alone doesn't know the partition ID, we set
it here
+
partitionData.partitionResponse.setPartition(entry.getKey().partition());
+ // We have to keep the order of input topic-partition. Hence, we
batch the partitions only if the last
+ // batch is in the same topic group.
+ if (!topicResponseList.isEmpty() &&
topicResponseList.get(topicResponseList.size() - 1)
+ .topic().equals(entry.getKey().topic())) {
+ topicResponseList.get(topicResponseList.size() - 1)
Review comment:
Maybe we could have a variable called `previousTopic` to make this code
more readable.
----------------------------------------------------------------
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:
[email protected]
