/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.connect.util;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.CreateTopicsOptions;
import org.apache.kafka.clients.admin.DescribeConfigsOptions;
import org.apache.kafka.clients.admin.DescribeTopicsOptions;
import org.apache.kafka.clients.admin.ListOffsetsOptions;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.ClusterAuthorizationException;
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Timer;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.RetriableException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Utility to simplify creating and managing topics via the {@link Admin}.
 */
public class TopicAdmin implements AutoCloseable {

    public static final TopicCreationResponse EMPTY_CREATION = new TopicCreationResponse(Collections.emptySet(), Collections.emptySet());
    private static final List<Class<? extends Exception>> CAUSES_TO_RETRY_TOPIC_CREATION = Arrays.asList(
            InvalidReplicationFactorException.class,
            TimeoutException.class);

    public static class TopicCreationResponse {

        private final Set<String> created;
        private final Set<String> existing;

        public TopicCreationResponse(Set<String> createdTopicNames, Set<String> existingTopicNames) {
            this.created = Collections.unmodifiableSet(createdTopicNames);
            this.existing = Collections.unmodifiableSet(existingTopicNames);
        }

        public Set<String> createdTopics() {
            return created;
        }

        public boolean isCreated(String topicName) {
            return created.contains(topicName);
        }

        public boolean isExisting(String topicName) {
            return existing.contains(topicName);
        }

        public boolean isCreatedOrExisting(String topicName) {
            return isCreated(topicName) || isExisting(topicName);
        }

        public int createdTopicsCount() {
            return created.size();
        }

        public int existingTopicsCount() {
            return existing.size();
        }

        public int createdOrExistingTopicsCount() {
            return createdTopicsCount() + existingTopicsCount();
        }

        public boolean isEmpty() {
            return createdOrExistingTopicsCount() == 0;
        }

        @Override
        public String toString() {
            return "TopicCreationResponse{" + "created=" + created + ", existing=" + existing + '}';
        }
    }

    public static final int NO_PARTITIONS = -1;
    public static final short NO_REPLICATION_FACTOR = -1;

    private static final String CLEANUP_POLICY_CONFIG = TopicConfig.CLEANUP_POLICY_CONFIG;
    private static final String CLEANUP_POLICY_COMPACT = TopicConfig.CLEANUP_POLICY_COMPACT;

    /**
     * A builder of {@link NewTopic} instances.
     */
    public static class NewTopicBuilder {
        private final String name;
        private int numPartitions = NO_PARTITIONS;
        private short replicationFactor = NO_REPLICATION_FACTOR;
        private final Map<String, String> configs = new HashMap<>();

        NewTopicBuilder(String name) {
            this.name = name;
        }

        /**
         * Specify the desired number of partitions for the topic.
         *
         * @param numPartitions the desired number of partitions; must be positive, or -1 to
         *                      signify using the broker's default
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder partitions(int numPartitions) {
            this.numPartitions = numPartitions;
            return this;
        }

        /**
         * Specify the topic's number of partition should be the broker configuration for
         * {@code num.partitions}.
         *
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder defaultPartitions() {
            this.numPartitions = NO_PARTITIONS;
            return this;
        }

        /**
         * Specify the desired replication factor for the topic.
         *
         * @param replicationFactor the desired replication factor; must be positive, or -1 to
         *                          signify using the broker's default
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder replicationFactor(short replicationFactor) {
            this.replicationFactor = replicationFactor;
            return this;
        }

        /**
         * Specify the replication factor for the topic should be the broker configurations for
         * {@code default.replication.factor}.
         *
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder defaultReplicationFactor() {
            this.replicationFactor = NO_REPLICATION_FACTOR;
            return this;
        }

        /**
         * Specify that the topic should be compacted.
         *
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder compacted() {
            this.configs.put(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT);
            return this;
        }

        /**
         * Specify the configuration properties for the topic, overwriting any previously-set properties.
         *
         * @param configs the desired topic configuration properties, or null if all existing properties should be cleared
         * @return this builder to allow methods to be chained; never null
         */
        public NewTopicBuilder config(Map<String, Object> configs) {
            if (configs != null) {
                for (Map.Entry<String, Object> entry : configs.entrySet()) {
                    Object value = entry.getValue();
                    this.configs.put(entry.getKey(), value != null ? value.toString() : null);
                }
            } else {
                this.configs.clear();
            }
            return this;
        }

        /**
         * Build the {@link NewTopic} representation.
         *
         * @return the topic description; never null
         */
        public NewTopic build() {
            return new NewTopic(
                    name,
                    Optional.of(numPartitions),
                    Optional.of(replicationFactor)
            ).configs(configs);
        }
    }

    /**
     * Obtain a {@link NewTopicBuilder builder} to define a {@link NewTopic}.
     *
     * @param topicName the name of the topic
     * @return the {@link NewTopic} description of the topic; never null
     */
    public static NewTopicBuilder defineTopic(String topicName) {
        return new NewTopicBuilder(topicName);
    }

    private static final Logger log = LoggerFactory.getLogger(TopicAdmin.class);
    private final String bootstrapServers;
    private final Admin admin;
    private final boolean logCreation;

    /**
     * Create a new topic admin component with the given configuration.
     * <p>
     * Note that this will create an underlying {@link Admin} instance which must be freed when this
     * topic admin is no longer needed by calling {@link #close()} or {@link #close(Duration)}.
     *
     * @param adminConfig the configuration for the {@link Admin}
     */
    public TopicAdmin(Map<String, Object> adminConfig) {
        this(adminConfig, Admin.create(adminConfig));
    }

    public TopicAdmin(Map<String, Object> adminConfig, Admin adminClient) {
        this(bootstrapServers(adminConfig), adminClient, true);
    }

    // visible for testing
    TopicAdmin(Admin adminClient) {
        this(null, adminClient, true);
    }

    // visible for testing
    TopicAdmin(String bootstrapServers, Admin adminClient, boolean logCreation) {
        this.admin = adminClient;
        this.bootstrapServers = bootstrapServers != null ? bootstrapServers : "<unknown>";
        this.logCreation = logCreation;
    }

    private static String bootstrapServers(Map<String, Object> adminConfig) {
        Object result = adminConfig.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG);
        return result != null ? result.toString() : null;
    }

   /**
     * Attempt to create the topic described by the given definition, returning true if the topic was created or false
     * if the topic already existed.
     *
     * @param topic the specification of the topic
     * @return true if the topic was created or false if the topic already existed.
     * @throws ConnectException            if an error occurs, the operation takes too long, or the thread is interrupted while
     *                                     attempting to perform this operation
     * @throws UnsupportedVersionException if the broker does not support the necessary APIs to perform this request
     */
    public boolean createTopic(NewTopic topic) {
        if (topic == null) return false;
        Set<String> newTopicNames = createTopics(topic);
        return newTopicNames.contains(topic.name());
    }

    /**
     * Attempt to create the topics described by the given definitions, returning all of the names of those topics that
     * were created by this request. Any existing topics with the same name are unchanged, and the names of such topics
     * are excluded from the result.
     * <p>
     * If multiple topic definitions have the same topic name, the last one with that name will be used.
     * <p>
     * Apache Kafka added support for creating topics in 0.10.1.0, so this method works as expected with that and later versions.
     * With brokers older than 0.10.1.0, this method is unable to create topics and always returns an empty set.
     *
     * @param topics the specifications of the topics
     * @return the names of the topics that were created by this operation; never null but possibly empty
     * @throws ConnectException            if an error occurs, the operation takes too long, or the thread is interrupted while
     *                                     attempting to perform this operation
     */
    public Set<String> createTopics(NewTopic... topics) {
        return createOrFindTopics(topics).createdTopics();
    }

    /**
     * Implements a retry logic around creating topic(s) in case it'd fail due to
     * specific type of exceptions, see {@link TopicAdmin#retryableTopicCreationException(ConnectException)}
     *
     * @param topicDescription the specifications of the topic
     * @param timeoutMs        Timeout in milliseconds
     * @param backOffMs        Time for delay after initial failed attempt in milliseconds
     * @param time             {@link Time} instance
     * @return the names of the topics that were created by this operation; never null but possibly empty,
     * the same as {@link TopicAdmin#createTopics(NewTopic...)}
     */
    public Set<String> createTopicsWithRetry(NewTopic topicDescription, long timeoutMs, long backOffMs, Time time) {
        Timer timer = time.timer(timeoutMs);
        do {
            try {
                return createTopics(topicDescription);
            } catch (ConnectException e) {
                if (timer.notExpired() && retryableTopicCreationException(e)) {
                    log.info("'{}' topic creation failed due to '{}', retrying, {}ms remaining",
                            topicDescription.name(), e.getMessage(), timer.remainingMs());
                } else {
                    throw e;
                }
            }
            timer.sleep(backOffMs);
        } while (timer.notExpired());
        throw new TimeoutException("Timeout expired while trying to create topic(s)");
    }

    private boolean retryableTopicCreationException(ConnectException e) {
        // createTopics wraps the exception into ConnectException
        // to retry the creation, it should be an ExecutionException from future get which was caused by InvalidReplicationFactorException
        // or can be a TimeoutException
        Throwable cause = e.getCause();
        while (cause != null) {
            final Throwable finalCause = cause;
            if (CAUSES_TO_RETRY_TOPIC_CREATION.stream().anyMatch(exceptionClass -> exceptionClass.isInstance(finalCause))) {
                return true;
            }
            cause = cause.getCause();
        }
        return false;
    }

    /**
     * Attempt to find or create the topic described by the given definition, returning true if the topic was created or had
     * already existed, or false if the topic did not exist and could not be created.
     *
     * @param topic the specification of the topic
     * @return true if the topic was created or existed, or false if the topic could not already existed.
     * @throws ConnectException            if an error occurs, the operation takes too long, or the thread is interrupted while
     *                                     attempting to perform this operation
     * @throws UnsupportedVersionException if the broker does not support the necessary APIs to perform this request
     */
    public boolean createOrFindTopic(NewTopic topic) {
        if (topic == null) return false;
        return createOrFindTopics(topic).isCreatedOrExisting(topic.name());
    }

    /**
     * Attempt to create the topics described by the given definitions, returning all of the names of those topics that
     * were created by this request. Any existing topics with the same name are unchanged, and the names of such topics
     * are excluded from the result.
     * <p>
     * If multiple topic definitions have the same topic name, the last one with that name will be used.
     * <p>
     * Apache Kafka added support for creating topics in 0.10.1.0, so this method works as expected with that and later versions.
     * With brokers older than 0.10.1.0, this method is unable to create topics and always returns an empty set.
     *
     * @param topics the specifications of the topics
     * @return the {@link TopicCreationResponse} with the names of the newly created and existing topics;
     *         never null but possibly empty
     * @throws ConnectException if an error occurs, the operation takes too long, or the thread is interrupted while
     *                          attempting to perform this operation
     */
    public TopicCreationResponse createOrFindTopics(NewTopic... topics) {
        Map<String, NewTopic> topicsByName = new HashMap<>();
        if (topics != null) {
            for (NewTopic topic : topics) {
                if (topic != null) topicsByName.put(topic.name(), topic);
            }
        }
        if (topicsByName.isEmpty()) return EMPTY_CREATION;
        String topicNameList = String.join("', '", topicsByName.keySet());

        // Attempt to create any missing topics
        CreateTopicsOptions args = new CreateTopicsOptions().validateOnly(false);
        Map<String, KafkaFuture<Void>> newResults = admin.createTopics(topicsByName.values(), args).values();

        // Iterate over each future so that we can handle individual failures like when some topics already exist
        Set<String> newlyCreatedTopicNames = new HashSet<>();
        Set<String> existingTopicNames = new HashSet<>();
        for (Map.Entry<String, KafkaFuture<Void>> entry : newResults.entrySet()) {
            String topic = entry.getKey();
            try {
                entry.getValue().get();
                if (logCreation) {
                    log.info("Created topic {} on brokers at {}", topicsByName.get(topic), bootstrapServers);
                }
                newlyCreatedTopicNames.add(topic);
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof TopicExistsException) {
                    log.debug("Found existing topic '{}' on the brokers at {}", topic, bootstrapServers);
                    existingTopicNames.add(topic);
                    continue;
                }
                if (cause instanceof UnsupportedVersionException) {
                    log.debug("Unable to create topic(s) '{}' since the brokers at {} do not support the CreateTopics API." +
                            " Falling back to assume topic(s) exist or will be auto-created by the broker.",
                            topicNameList, bootstrapServers);
                    return EMPTY_CREATION;
                }
                if (cause instanceof ClusterAuthorizationException) {
                    log.debug("Not authorized to create topic(s) '{}' upon the brokers {}." +
                            " Falling back to assume topic(s) exist or will be auto-created by the broker.",
                            topicNameList, bootstrapServers);
                    return EMPTY_CREATION;
                }
                if (cause instanceof TopicAuthorizationException) {
                    log.debug("Not authorized to create topic(s) '{}' upon the brokers {}." +
                                    " Falling back to assume topic(s) exist or will be auto-created by the broker.",
                            topicNameList, bootstrapServers);
                    return EMPTY_CREATION;
                }
                if (cause instanceof InvalidConfigurationException) {
                    throw new ConnectException("Unable to create topic(s) '" + topicNameList + "': " + cause.getMessage(),
                            cause);
                }
                if (cause instanceof TimeoutException) {
                    // Timed out waiting for the operation to complete
                    throw new ConnectException("Timed out while checking for or creating topic(s) '" + topicNameList + "'." +
                            " This could indicate a connectivity issue, unavailable topic partitions, or if" +
                            " this is your first use of the topic it may have taken too long to create.", cause);
                }
                throw new ConnectException("Error while attempting to create/find topic(s) '" + topicNameList + "'", e);
            } catch (InterruptedException e) {
                Thread.interrupted();
                throw new ConnectException("Interrupted while attempting to create/find topic(s) '" + topicNameList + "'", e);
            }
        }
        return new TopicCreationResponse(newlyCreatedTopicNames, existingTopicNames);
    }

    /**
     * Attempt to fetch the descriptions of the given topics
     * Apache Kafka added support for describing topics in 0.10.0.0, so this method works as expected with that and later versions.
     * With brokers older than 0.10.0.0, this method is unable to describe topics and always returns an empty set.
     *
     * @param topics the topics to describe
     * @return a map of topic names to topic descriptions of the topics that were requested; never null but possibly empty
     * @throws RetriableException if a retriable error occurs, the operation takes too long, or the
     * thread is interrupted while attempting to perform this operation
     * @throws ConnectException if a non retriable error occurs
     */
    public Map<String, TopicDescription> describeTopics(String... topics) {
        if (topics == null) {
            return Collections.emptyMap();
        }
        String topicNameList = String.join(", ", topics);

        Map<String, KafkaFuture<TopicDescription>> newResults =
                admin.describeTopics(Arrays.asList(topics), new DescribeTopicsOptions()).topicNameValues();

        // Iterate over each future so that we can handle individual failures like when some topics don't exist
        Map<String, TopicDescription> existingTopics = new HashMap<>();
        newResults.forEach((topic, desc) -> {
            try {
                existingTopics.put(topic, desc.get());
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof UnknownTopicOrPartitionException) {
                    log.debug("Topic '{}' does not exist on the brokers at {}", topic, bootstrapServers);
                    return;
                }
                if (cause instanceof ClusterAuthorizationException || cause instanceof TopicAuthorizationException) {
                    String msg = String.format("Not authorized to describe topic(s) '%s' on the brokers %s",
                            topicNameList, bootstrapServers);
                    throw new ConnectException(msg, cause);
                }
                if (cause instanceof UnsupportedVersionException) {
                    String msg = String.format("Unable to describe topic(s) '%s' since the brokers "
                                    + "at %s do not support the DescribeTopics API.",
                            topicNameList, bootstrapServers);
                    throw new ConnectException(msg, cause);
                }
                if (cause instanceof TimeoutException) {
                    // Timed out waiting for the operation to complete
                    throw new RetriableException("Timed out while describing topics '" + topicNameList + "'", cause);
                }
                throw new ConnectException("Error while attempting to describe topics '" + topicNameList + "'", e);
            } catch (InterruptedException e) {
                Thread.interrupted();
                throw new RetriableException("Interrupted while attempting to describe topics '" + topicNameList + "'", e);
            }
        });
        return existingTopics;
    }

    /**
     * Verify the named topic uses only compaction for the cleanup policy.
     *
     * @param topic             the name of the topic
     * @param workerTopicConfig the name of the worker configuration that specifies the topic name
     * @return true if the admin client could be used to verify the topic setting, or false if
     *         the verification could not be performed, likely because the admin client principal
     *         did not have the required permissions or because the broker was older than 0.11.0.0
     * @throws ConfigException if the actual topic setting did not match the required setting
     */
    public boolean verifyTopicCleanupPolicyOnlyCompact(String topic, String workerTopicConfig,
            String topicPurpose) {
        Set<String> cleanupPolicies = topicCleanupPolicy(topic);
        if (cleanupPolicies.isEmpty()) {
            log.info("Unable to use admin client to verify the cleanup policy of '{}' "
                      + "topic is '{}', either because the broker is an older "
                      + "version or because the Kafka principal used for Connect "
                      + "internal topics does not have the required permission to "
                      + "describe topic configurations.", topic, TopicConfig.CLEANUP_POLICY_COMPACT);
            return false;
        }
        Set<String> expectedPolicies = Collections.singleton(TopicConfig.CLEANUP_POLICY_COMPACT);
        if (!cleanupPolicies.equals(expectedPolicies)) {
            String expectedPolicyStr = String.join(",", expectedPolicies);
            String cleanupPolicyStr = String.join(",", cleanupPolicies);
            String msg = String.format("Topic '%s' supplied via the '%s' property is required "
                    + "to have '%s=%s' to guarantee consistency and durability of "
                    + "%s, but found the topic currently has '%s=%s'. Continuing would likely "
                    + "result in eventually losing %s and problems restarting this Connect "
                    + "cluster in the future. Change the '%s' property in the "
                    + "Connect worker configurations to use a topic with '%s=%s'.",
                    topic, workerTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, expectedPolicyStr,
                    topicPurpose, TopicConfig.CLEANUP_POLICY_CONFIG, cleanupPolicyStr, topicPurpose,
                    workerTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, expectedPolicyStr);
            throw new ConfigException(msg);
        }
        return true;
    }

    /**
     * Get the cleanup policy for a topic.
     *
     * @param topic the name of the topic
     * @return the set of cleanup policies set for the topic; may be empty if the topic does not
     *         exist or the topic's cleanup policy could not be retrieved
     */
    public Set<String> topicCleanupPolicy(String topic) {
        Config topicConfig = describeTopicConfig(topic);
        if (topicConfig == null) {
            // The topic must not exist
            log.debug("Unable to find topic '{}' when getting cleanup policy", topic);
            return Collections.emptySet();
        }
        ConfigEntry entry = topicConfig.get(CLEANUP_POLICY_CONFIG);
        if (entry != null && entry.value() != null) {
            String policyStr = entry.value();
            log.debug("Found cleanup.policy={} for topic '{}'", policyStr, topic);
            return Arrays.stream(policyStr.split(","))
                         .map(String::trim)
                         .filter(s -> !s.isEmpty())
                         .map(String::toLowerCase)
                         .collect(Collectors.toSet());
        }
        // This is unexpected, as the topic config should include the cleanup.policy even if
        // the topic settings don't override the broker's log.cleanup.policy. But just to be safe.
        log.debug("Found no cleanup.policy for topic '{}'", topic);
        return Collections.emptySet();
    }

    /**
     * Attempt to fetch the topic configuration for the given topic.
     * Apache Kafka added support for describing topic configurations in 0.11.0.0, so this method
     * works as expected with that and later versions. With brokers older than 0.11.0.0, this method
     * is unable get the topic configurations and always returns a null value.
     *
     * <p>If the topic does not exist, a null value is returned.
     *
     * @param topic the name of the topic for which the topic configuration should be obtained
     * @return the topic configuration if the topic exists, or null if the topic did not exist
     * @throws RetriableException if a retriable error occurs, the operation takes too long, or the
     *         thread is interrupted while attempting to perform this operation
     * @throws ConnectException if a non retriable error occurs
     */
    public Config describeTopicConfig(String topic) {
        return describeTopicConfigs(topic).get(topic);
    }

    /**
     * Attempt to fetch the topic configurations for the given topics.
     * Apache Kafka added support for describing topic configurations in 0.11.0.0, so this method
     * works as expected with that and later versions. With brokers older than 0.11.0.0, this method
     * is unable get the topic configurations and always returns an empty set.
     *
     * <p>An entry with a null Config is placed into the resulting map for any topic that does
     * not exist on the brokers.
     *
     * @param topicNames the topics to obtain configurations
     * @return the map of topic configurations for each existing topic, or an empty map if none
     * of the topics exist
     * @throws RetriableException if a retriable error occurs, the operation takes too long, or the
     *         thread is interrupted while attempting to perform this operation
     * @throws ConnectException if a non retriable error occurs
     */
    public Map<String, Config> describeTopicConfigs(String... topicNames) {
        if (topicNames == null) {
            return Collections.emptyMap();
        }
        Collection<String> topics = Arrays.stream(topicNames)
                                          .filter(Objects::nonNull)
                                          .map(String::trim)
                                          .filter(s -> !s.isEmpty())
                                          .collect(Collectors.toList());
        if (topics.isEmpty()) {
            return Collections.emptyMap();
        }
        String topicNameList = String.join(", ", topics);
        Collection<ConfigResource> resources = topics.stream()
                                                     .map(t -> new ConfigResource(ConfigResource.Type.TOPIC, t))
                                                     .collect(Collectors.toList());

        Map<ConfigResource, KafkaFuture<Config>> newResults = admin.describeConfigs(resources, new DescribeConfigsOptions()).values();

        // Iterate over each future so that we can handle individual failures like when some topics don't exist
        Map<String, Config> result = new HashMap<>();
        newResults.forEach((resource, configs) -> {
            String topic = resource.name();
            try {
                result.put(topic, configs.get());
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof UnknownTopicOrPartitionException) {
                    log.debug("Topic '{}' does not exist on the brokers at {}", topic, bootstrapServers);
                    result.put(topic, null);
                } else if (cause instanceof ClusterAuthorizationException || cause instanceof TopicAuthorizationException) {
                    log.debug("Not authorized to describe topic config for topic '{}' on brokers at {}", topic, bootstrapServers);
                } else if (cause instanceof UnsupportedVersionException) {
                    log.debug("API to describe topic config for topic '{}' is unsupported on brokers at {}", topic, bootstrapServers);
                } else if (cause instanceof TimeoutException) {
                    String msg = String.format("Timed out while waiting to describe topic config for topic '%s' on brokers at %s",
                            topic, bootstrapServers);
                    throw new RetriableException(msg, e);
                } else {
                    String msg = String.format("Error while attempting to describe topic config for topic '%s' on brokers at %s",
                            topic, bootstrapServers);
                    throw new ConnectException(msg, e);
                }
            } catch (InterruptedException e) {
                Thread.interrupted();
                String msg = String.format("Interrupted while attempting to describe topic configs '%s'", topicNameList);
                throw new RetriableException(msg, e);
            }
        });
        return result;
    }

    /**
     * Fetch the most recent offset for each of the supplied {@link TopicPartition} objects.
     *
     * @param partitions the topic partitions
     * @return the map of offset for each topic partition, or an empty map if the supplied partitions
     *         are null or empty
     * @throws UnsupportedVersionException if the admin client cannot read end offsets
     * @throws TimeoutException if the offset metadata could not be fetched before the amount of time allocated
     *         by {@code request.timeout.ms} expires, and this call can be retried
     * @throws LeaderNotAvailableException if the leader was not available and this call can be retried
     * @throws RetriableException if a retriable error occurs, or the thread is interrupted while attempting
     *         to perform this operation
     * @throws ConnectException if a non retriable error occurs
     */
    public Map<TopicPartition, Long> endOffsets(Set<TopicPartition> partitions) {
        if (partitions == null || partitions.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<TopicPartition, OffsetSpec> offsetSpecMap = partitions.stream().collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest()));
        ListOffsetsResult resultFuture = admin.listOffsets(offsetSpecMap, new ListOffsetsOptions(IsolationLevel.READ_UNCOMMITTED));
        // Get the individual result for each topic partition so we have better error messages
        Map<TopicPartition, Long> result = new HashMap<>();
        for (TopicPartition partition : partitions) {
            try {
                ListOffsetsResultInfo info = resultFuture.partitionResult(partition).get();
                result.put(partition, info.offset());
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                String topic = partition.topic();
                if (cause instanceof AuthorizationException) {
                    String msg = String.format("Not authorized to get the end offsets for topic '%s' on brokers at %s", topic, bootstrapServers);
                    throw new ConnectException(msg, cause);
                } else if (cause instanceof UnsupportedVersionException) {
                    // Should theoretically never happen, because this method is the same as what the consumer uses and therefore
                    // should exist in the broker since before the admin client was added
                    String msg = String.format("API to get the get the end offsets for topic '%s' is unsupported on brokers at %s", topic, bootstrapServers);
                    throw new UnsupportedVersionException(msg, cause);
                } else if (cause instanceof TimeoutException) {
                    String msg = String.format("Timed out while waiting to get end offsets for topic '%s' on brokers at %s", topic, bootstrapServers);
                    throw new TimeoutException(msg, cause);
                } else if (cause instanceof LeaderNotAvailableException) {
                    String msg = String.format("Unable to get end offsets during leader election for topic '%s' on brokers at %s", topic, bootstrapServers);
                    throw new LeaderNotAvailableException(msg, cause);
                } else if (cause instanceof org.apache.kafka.common.errors.RetriableException) {
                    throw (org.apache.kafka.common.errors.RetriableException) cause;
                } else {
                    String msg = String.format("Error while getting end offsets for topic '%s' on brokers at %s", topic, bootstrapServers);
                    throw new ConnectException(msg, cause);
                }
            } catch (InterruptedException e) {
                Thread.interrupted();
                String msg = String.format("Interrupted while attempting to read end offsets for topic '%s' on brokers at %s", partition.topic(), bootstrapServers);
                throw new RetriableException(msg, e);
            }
        }
        return result;
    }

    /**
     * Fetch the most recent offset for each of the supplied {@link TopicPartition} objects, and performs retry when
     * {@link org.apache.kafka.connect.errors.RetriableException} is thrown.
     *
     * @param partitions        the topic partitions
     * @param timeoutDuration   timeout duration; may not be null
     * @param retryBackoffMs    the number of milliseconds to delay upon receiving a
     *                          {@link org.apache.kafka.connect.errors.RetriableException} before retrying again;
     *                          must be 0 or more
     * @return                  the map of offset for each topic partition, or an empty map if the supplied partitions
     *                          are null or empty
     * @throws UnsupportedVersionException if the broker is too old to support the admin client API to read end offsets
     * @throws ConnectException if {@code timeoutDuration} is exhausted
     * @see TopicAdmin#endOffsets(Set)
     */
    public Map<TopicPartition, Long> retryEndOffsets(Set<TopicPartition> partitions, Duration timeoutDuration, long retryBackoffMs) {

        try {
            return RetryUtil.retryUntilTimeout(
                    () -> endOffsets(partitions),
                    () -> "list offsets for topic partitions",
                    timeoutDuration,
                    retryBackoffMs);
        } catch (UnsupportedVersionException e) {
            // Older brokers don't support this admin method, so rethrow it without wrapping it
            throw e;
        } catch (Exception e) {
            throw ConnectUtils.maybeWrap(e, "Failed to list offsets for topic partitions");
        }
    }

    @Override
    public void close() {
        admin.close();
    }

    public void close(Duration timeout) {
        admin.close(timeout);
    }
}
