private static FetchResponseData toMessage(Errors error,
                                               int throttleTimeMs,
                                               int sessionId,
                                               Iterator<Map.Entry<TopicIdPartition, FetchResponseData.PartitionData>> partIterator,
                                               List<Node> nodeEndpoints) {
        List<FetchResponseData.FetchableTopicResponse> topicResponseList = new ArrayList<>();
        while (partIterator.hasNext()) {
            Map.Entry<TopicIdPartition, FetchResponseData.PartitionData> entry = partIterator.next();
            FetchResponseData.PartitionData partitionData = entry.getValue();
            // Since PartitionData alone doesn't know the partition ID, we set it here
            partitionData.setPartitionIndex(entry.getKey().topicPartition().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.
            FetchResponseData.FetchableTopicResponse previousTopic = topicResponseList.isEmpty() ? null
                : topicResponseList.get(topicResponseList.size() - 1);
            if (matchingTopic(previousTopic, entry.getKey()))
                previousTopic.partitions().add(partitionData);
            else {
                List<FetchResponseData.PartitionData> partitionResponses = new ArrayList<>();
                partitionResponses.add(partitionData);
                topicResponseList.add(new FetchResponseData.FetchableTopicResponse()
                    .setTopic(entry.getKey().topicPartition().topic())
                    .setTopicId(entry.getKey().topicId())
                    .setPartitions(partitionResponses));
            }
        }
        FetchResponseData data = new FetchResponseData();
        // KafkaApis should only pass in node endpoints on error, otherwise this should be an empty list
        nodeEndpoints.forEach(endpoint -> data.nodeEndpoints().add(
                new FetchResponseData.NodeEndpoint()
                        .setNodeId(endpoint.id())
                        .setHost(endpoint.host())
                        .setPort(endpoint.port())
                        .setRack(endpoint.rack())));
        return data.setThrottleTimeMs(throttleTimeMs)
                .setErrorCode(error.code())
                .setSessionId(sessionId)
                .setResponses(topicResponseList);
    }
@return The size in bytes of the records. 0 is returned if records of input partition is null.
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java