/**
     * Pause specific partitions for a consumer group in the cluster, with the default options.
     * <p>
     * This is a convenience method for {@link #pausePartitions(String, Set, PausePartitionsOptions)}
     * with default options. See the overload for more details.
     *
     * @param groupId    The ID of the consumer group.
     * @param partitions The partitions to pause.
     * @return The PausePartitionsResult.
     */
    default PausePartitionsResult pausePartitions(String groupId, Set<TopicPartition> partitions) {
        return pausePartitions(groupId, partitions, new PausePartitionsOptions());
    }

    /**
     * Pause specific partitions for a consumer group in the cluster.
     *
     * @param groupId    The ID of the consumer group.
     * @param partitions The partitions to pause.
     * @param options    The options to use when pausing the partitions.
     * @return The PausePartitionsResult.
     */
    PausePartitionsResult pausePartitions(String groupId,
                                          Set<TopicPartition> partitions,
                                          PausePartitionsOptions options);

    /**
     * Resume specific paused partitions for a consumer group in the cluster, with the default options.
     * <p>
     * This is a convenience method for {@link #resumePartitions(String, Set, ResumePartitionsOptions)}
     * with default options. See the overload for more details.
     *
     * @param groupId    The ID of the consumer group.
     * @param partitions The partitions to resume.
     * @return The ResumePartitionsResult.
     */
    default ResumePartitionsResult resumePartitions(String groupId, Set<TopicPartition> partitions) {
        return resumePartitions(groupId, partitions, new ResumePartitionsOptions());
    }

    /**
     * Resume specific paused partitions for a consumer group in the cluster.
     *
     * @param groupId    The ID of the consumer group.
     * @param partitions The partitions to resume.
     * @param options    The options to use when resuming the partitions.
     * @return The ResumePartitionsResult.
     */
    ResumePartitionsResult resumePartitions(String groupId,
                                            Set<TopicPartition> partitions,
                                            ResumePartitionsOptions options);

    /**
     * List the paused partitions for a consumer group in the cluster, with the default options.
     * <p>
     * This is a convenience method for {@link #listPausedPartitions(String, ListPausedPartitionsOptions)}
     * with default options. See the overload for more details.
     *
     * @param groupId The ID of the consumer group.
     * @return The ListPausedPartitionsResult.
     */
    default ListPausedPartitionsResult listPausedPartitions(String groupId) {
        return listPausedPartitions(groupId, new ListPausedPartitionsOptions());
    }

    /**
     * List the paused partitions for a consumer group in the cluster.
     *
     * @param groupId The ID of the consumer group.
     * @param options The options to use when describing the paused partitions.
     * @return The ListPausedPartitionsResult.
     */
    ListPausedPartitionsResult listPausedPartitions(String groupId, ListPausedPartitionsOptions options);

/**
 * The result of {@link Admin#pausePartitions(String, Set, PausePartitionsOptions)} .
 */
public class PausePartitionsResult {

    private final KafkaFuture<Map<TopicPartition, Errors>> future;

    PausePartitionsResult(KafkaFuture<Map<TopicPartition, Errors>> future) {
        this.future = future;
    }

    /**
     * Return a future which can be used to check the result for a given partition.
     */
    public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();

        this.future.whenComplete((topicPartitions, throwable) -> {
            if (throwable != null) {
                result.completeExceptionally(throwable);
            } else if (!topicPartitions.containsKey(partition)) {
                result.completeExceptionally(new IllegalArgumentException(
                    "Pausing of the \"" + partition + "\" was not attempted"));
            } else {
                final Errors error = topicPartitions.get(partition);
                if (error == Errors.NONE) {
                    result.complete(null);
                } else {
                    result.completeExceptionally(error.exception());
                }
            }
        });

        return result;
    }

    /**
     * Return a future which succeeds if pausing of all the partitions succeed.
     */
    public KafkaFuture<Void> all() {
        return this.future.thenApply(topicPartitionErrorsMap ->  {
            List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
                                                                           .stream()
                                                                           .filter(e -> e.getValue() != Errors.NONE)
                                                                           .map(Map.Entry::getKey)
                                                                           .collect(Collectors.toList());
            for (Errors error : topicPartitionErrorsMap.values()) {
                if (error != Errors.NONE) {
                    throw error.exception(
                        "Failed pausing the following partitions: " + partitionsFailed);
                }
            }
            return null;
        });
    }
}

/**
 * Options for {@link Admin#pausePartitions(String, Set, PausePartitionsOptions)}.
 */
public class PausePartitionsOptions extends AbstractOptions<PausePartitionsOptions> {
}

/**
 * The result of {@link Admin#resumePartitions(String, Set, ResumePartitionsOptions)} .
 */
public class ResumePartitionsResult {

    private final KafkaFuture<Map<TopicPartition, Errors>> future;

    ResumePartitionsResult(KafkaFuture<Map<TopicPartition, Errors>> future) {
        this.future = future;
    }

    /**
     * Return a future which can be used to check the result for a given partition.
     */
    public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();

        this.future.whenComplete((topicPartitions, throwable) -> {
            if (throwable != null) {
                result.completeExceptionally(throwable);
            } else if (!topicPartitions.containsKey(partition)) {
                result.completeExceptionally(new IllegalArgumentException(
                    "Resuming of the \"" + partition + "\" was not attempted"));
            } else {
                final Errors error = topicPartitions.get(partition);
                if (error == Errors.NONE) {
                    result.complete(null);
                } else {
                    result.completeExceptionally(error.exception());
                }
            }
        });

        return result;
    }

    /**
     * Return a future which succeeds if resuming of all the partitions succeed.
     */
    public KafkaFuture<Void> all() {
        return this.future.thenApply(topicPartitionErrorsMap ->  {
            List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
                                                                           .stream()
                                                                           .filter(e -> e.getValue() != Errors.NONE)
                                                                           .map(Map.Entry::getKey)
                                                                           .collect(Collectors.toList());
            for (Errors error : topicPartitionErrorsMap.values()) {
                if (error != Errors.NONE) {
                    throw error.exception(
                        "Failed resuming the following partitions: " + partitionsFailed);
                }
            }
            return null;
        });
    }
}

/**
 * Options for {@link Admin#resumePartitions(String, Set, ResumePartitionsOptions)}.
 */
public class ResumePartitionsOptions extends AbstractOptions<ResumePartitionsOptions> {
}

/**
 * The result of {@link Admin#listPausedPartitions(String, ListPausedPartitionsOptions)}.
 */
public class ListPausedPartitionsResult {

    private final KafkaFuture<Map<TopicPartition, Long>> future;

    ListPausedPartitionsResult(KafkaFuture<Map<TopicPartition, Long>> future) {
        this.future = future;
    }

    /**
     * Return a future which can be used to check the result for a given partition.
     */
    public KafkaFuture<Long> partitionResult(final TopicPartition partition) {
        final KafkaFutureImpl<Long> result = new KafkaFutureImpl<>();

        this.future.whenComplete((topicPartitions, throwable) -> {
            if (throwable != null) {
                result.completeExceptionally(throwable);
            } else if (!topicPartitions.containsKey(partition)) {
                result.completeExceptionally(new IllegalArgumentException(
                    "Describing paused partitions for the \"" + partition + "\" was not attempted"));
            } else {
                result.complete(topicPartitions.get(partition));
            }
        });

        return result;
    }

    /**
     * Return a future.
     */
    public KafkaFuture<Map<TopicPartition, Long>> all() {
        return future;
    }
}

/**
 * Options for {@link Admin#listPausedPartitions(String, ListPausedPartitionsOptions)}.
 */
public class ListPausedPartitionsOptions extends AbstractOptions<ListPausedPartitionsOptions> {
}