public static ShareAcknowledgeResponseData toMessage(Errors error, int throttleTimeMs,
                                                         Iterator<Map.Entry<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData>> partIterator,
                                                         List<Node> nodeEndpoints) {
        Map<Uuid, ShareAcknowledgeResponseData.ShareAcknowledgeTopicResponse> topicResponseList = new LinkedHashMap<>();
        while (partIterator.hasNext()) {
            Map.Entry<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> entry = partIterator.next();
            ShareAcknowledgeResponseData.PartitionData partitionData = entry.getValue();
            // Since PartitionData alone doesn't know the partition ID, we set it here
            partitionData.setPartitionIndex(entry.getKey().topicPartition().partition());
            // Checking if the topic is already present in the map
            if (topicResponseList.containsKey(entry.getKey().topicId())) {
                topicResponseList.get(entry.getKey().topicId()).partitions().add(partitionData);
            } else {
                List<ShareAcknowledgeResponseData.PartitionData> partitionResponses = new ArrayList<>();
                partitionResponses.add(partitionData);
                topicResponseList.put(entry.getKey().topicId(), new ShareAcknowledgeResponseData.ShareAcknowledgeTopicResponse()
                        .setTopicId(entry.getKey().topicId())
                        .setPartitions(partitionResponses));
            }
        }
        ShareAcknowledgeResponseData data = new ShareAcknowledgeResponseData();
        // KafkaApis should only pass in node endpoints on error, otherwise this should be an empty list
        nodeEndpoints.forEach(endpoint -> data.nodeEndpoints().add(
                new ShareAcknowledgeResponseData.NodeEndpoint()
                        .setNodeId(endpoint.id())
                        .setHost(endpoint.host())
                        .setPort(endpoint.port())
                        .setRack(endpoint.rack())));
        return data.setThrottleTimeMs(throttleTimeMs)
                .setErrorCode(error.code())
                .setResponses(new ArrayList<>(topicResponseList.values()));
    }
Possible error codes.
- {@link Errors#GROUP_AUTHORIZATION_FAILED}
- {@link Errors#TOPIC_AUTHORIZATION_FAILED}
- {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
- {@link Errors#NOT_LEADER_OR_FOLLOWER}
- {@link Errors#UNKNOWN_TOPIC_ID}
- {@link Errors#INVALID_RECORD_STATE}
- {@link Errors#KAFKA_STORAGE_ERROR}
- {@link Errors#INVALID_REQUEST}
- {@link Errors#UNKNOWN_SERVER_ERROR}
clients/src/main/java/org/apache/kafka/common/requests/ShareAcknowledgeResponse.java