@SuppressWarnings("NPathComplexity")
            @Override
            void handleResponse(AbstractResponse abstractResponse) {
                DescribeTopicPartitionsResponse response = (DescribeTopicPartitionsResponse) abstractResponse;
                DescribeTopicPartitionsResponseData.Cursor responseCursor = response.data().nextCursor();
                // The topicDescription for the cursor topic of the current batch.
                TopicDescription nextTopicDescription = null;

                for (DescribeTopicPartitionsResponseTopic topic : response.data().topics()) {
                    String topicName = topic.name();
                    Errors error = Errors.forCode(topic.errorCode());

                    KafkaFutureImpl<TopicDescription> future = topicFutures.get(topicName);

                    if (error != Errors.NONE) {
                        future.completeExceptionally(error.exception());
                        topicsRequests.remove(topicName);
                        if (responseCursor != null && responseCursor.topicName().equals(topicName)) {
                            responseCursor = null;
                        }
                        continue;
                    }

                    TopicDescription currentTopicDescription = getTopicDescriptionFromDescribeTopicsResponseTopic(topic, nodes, options.includeAuthorizedOperations());

                    if (partiallyFinishedTopicDescription != null && partiallyFinishedTopicDescription.name().equals(topicName)) {
                        // Add the partitions for the cursor topic of the previous batch.
                        partiallyFinishedTopicDescription.partitions().addAll(currentTopicDescription.partitions());
                        continue;
                    }

                    if (responseCursor != null && responseCursor.topicName().equals(topicName)) {
                        // In the same batch of result, it may need to handle the partitions for the previous cursor
                        // topic and the current cursor topic. Cache the result in the nextTopicDescription.
                        nextTopicDescription = currentTopicDescription;
                        continue;
                    }

                    topicsRequests.remove(topicName);
                    future.complete(currentTopicDescription);
                }

                if (partiallyFinishedTopicDescription != null &&
                        (responseCursor == null || !responseCursor.topicName().equals(partiallyFinishedTopicDescription.name()))) {
                    // We can't simply check nextTopicDescription != null here to close the partiallyFinishedTopicDescription.
                    // Because the responseCursor topic may not show in the response.
                    String topicName = partiallyFinishedTopicDescription.name();
                    topicFutures.get(topicName).complete(partiallyFinishedTopicDescription);
                    topicsRequests.remove(topicName);
                    partiallyFinishedTopicDescription = null;
                }
                if (nextTopicDescription != null) {
                    partiallyFinishedTopicDescription = nextTopicDescription;
                }

                if (!topicsRequests.isEmpty()) {
                    runnable.call(this, time.milliseconds());
                }
            }
Fail futures in the given Map which were retried due to exceeding quota. We propagate
the initial error back to the caller if the request timed out.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java