You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/04 16:07:41 UTC
[GitHub] [kafka] ijuma commented on a change in pull request #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
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:
users@infra.apache.org