/*
 * 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.streams.processor.internals;

import java.util.Map.Entry;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.ConfigResource.Type;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.TimeoutException;
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.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.internals.ClientUtils.QuietConsumerConfig;
import org.slf4j.Logger;

import java.util.ArrayList;
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.BiConsumer;
import java.util.stream.Collectors;



public class InternalTopicManager {
    private final static String BUG_ERROR_MESSAGE = "This indicates a bug. " +
        "Please report at https://issues.apache.org/jira/projects/KAFKA/issues or to the dev-mailing list (https://kafka.apache.org/contact).";
    private final static String INTERRUPTED_ERROR_MESSAGE = "Thread got interrupted. " + BUG_ERROR_MESSAGE;

    private final Logger log;

    private final Time time;
    private final Admin adminClient;

    private final short replicationFactor;
    private final long windowChangeLogAdditionalRetention;
    private final long retryBackOffMs;
    private final long retryTimeoutMs;

    private final Map<String, String> defaultTopicConfigs = new HashMap<>();

    public InternalTopicManager(final Time time,
                                final Admin adminClient,
                                final StreamsConfig streamsConfig) {
        this.time = time;
        this.adminClient = adminClient;

        final LogContext logContext = new LogContext(String.format("stream-thread [%s] ", Thread.currentThread().getName()));
        log = logContext.logger(getClass());

        replicationFactor = streamsConfig.getInt(StreamsConfig.REPLICATION_FACTOR_CONFIG).shortValue();
        windowChangeLogAdditionalRetention = streamsConfig.getLong(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG);
        retryBackOffMs = streamsConfig.getLong(StreamsConfig.RETRY_BACKOFF_MS_CONFIG);
        final Map<String, Object> consumerConfig = streamsConfig.getMainConsumerConfigs("dummy", "dummy", -1);
        // need to add mandatory configs; otherwise `QuietConsumerConfig` throws
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        retryTimeoutMs = new QuietConsumerConfig(consumerConfig).getInt(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG) / 2L;

        log.debug("Configs:" + Utils.NL +
            "\t{} = {}" + Utils.NL +
            "\t{} = {}",
            StreamsConfig.REPLICATION_FACTOR_CONFIG, replicationFactor,
            StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, windowChangeLogAdditionalRetention);

        for (final Map.Entry<String, Object> entry : streamsConfig.originalsWithPrefix(StreamsConfig.TOPIC_PREFIX).entrySet()) {
            if (entry.getValue() != null) {
                defaultTopicConfigs.put(entry.getKey(), entry.getValue().toString());
            }
        }
    }

    static class ValidationResult {
        private final Set<String> missingTopics = new HashSet<>();
        private final Map<String, List<String>> misconfigurationsForTopics = new HashMap<>();

        public void addMissingTopic(final String topic) {
            missingTopics.add(topic);
        }

        public Set<String> missingTopics() {
            return Collections.unmodifiableSet(missingTopics);
        }

        public void addMisconfiguration(final String topic, final String message) {
            misconfigurationsForTopics.computeIfAbsent(topic, ignored -> new ArrayList<>())
                .add(message);
        }

        public Map<String, List<String>> misconfigurationsForTopics() {
            return Collections.unmodifiableMap(misconfigurationsForTopics);
        }
    }

    /**
     * Validates the internal topics passed.
     *
     * The validation of the internal topics verifies if the topics:
     * - are missing on the brokers
     * - have the expected number of partitions
     * - have configured a clean-up policy that avoids data loss
     *
     * @param topicConfigs internal topics to validate
     *
     * @return validation results that contains
     *         - the set of missing internal topics on the brokers
     *         - descriptions of misconfigurations per topic
     */
    public ValidationResult validate(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to validate internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final ValidationResult validationResult = new ValidationResult();
        final Set<String> topicDescriptionsStillToValidate = new HashSet<>(topicConfigs.keySet());
        final Set<String> topicConfigsStillToValidate = new HashSet<>(topicConfigs.keySet());
        while (!topicDescriptionsStillToValidate.isEmpty() || !topicConfigsStillToValidate.isEmpty()) {
            Map<String, KafkaFuture<TopicDescription>> descriptionsForTopic = Collections.emptyMap();
            if (!topicDescriptionsStillToValidate.isEmpty()) {
                final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicDescriptionsStillToValidate);
                descriptionsForTopic = describeTopicsResult.topicNameValues();
            }
            Map<String, KafkaFuture<Config>> configsForTopic = Collections.emptyMap();
            if (!topicConfigsStillToValidate.isEmpty()) {
                final DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(
                    topicConfigsStillToValidate.stream()
                        .map(topic -> new ConfigResource(Type.TOPIC, topic))
                        .collect(Collectors.toSet())
                );
                configsForTopic = describeConfigsResult.values().entrySet().stream()
                    .collect(Collectors.toMap(entry -> entry.getKey().name(), Map.Entry::getValue));
            }

            while (!descriptionsForTopic.isEmpty() || !configsForTopic.isEmpty()) {
                if (!descriptionsForTopic.isEmpty()) {
                    doValidateTopic(
                        validationResult,
                        descriptionsForTopic,
                        topicConfigs,
                        topicDescriptionsStillToValidate,
                        (streamsSide, brokerSide) -> validatePartitionCount(validationResult, streamsSide, brokerSide)
                    );
                }
                if (!configsForTopic.isEmpty()) {
                    doValidateTopic(
                        validationResult,
                        configsForTopic,
                        topicConfigs,
                        topicConfigsStillToValidate,
                        (streamsSide, brokerSide) -> validateCleanupPolicy(validationResult, streamsSide, brokerSide)
                    );
                }

                maybeThrowTimeoutException(
                    Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate),
                    deadline,
                    String.format("Could not validate internal topics within %d milliseconds. " +
                        "This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs)
                );

                if (!descriptionsForTopic.isEmpty() || !configsForTopic.isEmpty()) {
                    Utils.sleep(100);
                }
            }

            maybeSleep(
                Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate),
                deadline,
                "validated"
            );
        }

        log.info("Completed validation of internal topics {}.", topicConfigs.keySet());
        return validationResult;
    }

    private <V> void doValidateTopic(final ValidationResult validationResult,
                                     final Map<String, KafkaFuture<V>> futuresForTopic,
                                     final Map<String, InternalTopicConfig> topicsConfigs,
                                     final Set<String> topicsStillToValidate,
                                     final BiConsumer<InternalTopicConfig, V> validator) {
        for (final String topicName : new HashSet<>(topicsStillToValidate)) {
            if (!futuresForTopic.containsKey(topicName)) {
                throw new IllegalStateException("Description results do not contain topics to validate. " + BUG_ERROR_MESSAGE);
            }
            final KafkaFuture<V> future = futuresForTopic.get(topicName);
            if (future.isDone()) {
                try {
                    final V brokerSideTopicConfig = future.get();
                    final InternalTopicConfig streamsSideTopicConfig = topicsConfigs.get(topicName);
                    validator.accept(streamsSideTopicConfig, brokerSideTopicConfig);
                    topicsStillToValidate.remove(topicName);
                } catch (final ExecutionException executionException) {
                    final Throwable cause = executionException.getCause();
                    if (cause instanceof UnknownTopicOrPartitionException) {
                        log.info("Internal topic {} is missing", topicName);
                        validationResult.addMissingTopic(topicName);
                        topicsStillToValidate.remove(topicName);
                    } else if (cause instanceof LeaderNotAvailableException) {
                        log.info("The leader of internal topic {} is not available.", topicName);
                    } else if (cause instanceof TimeoutException) {
                        log.info("Retrieving data for internal topic {} timed out.", topicName);
                    } else {
                        log.error("Unexpected error during internal topic validation: ", cause);
                        throw new StreamsException(
                            String.format("Could not validate internal topic %s for the following reason: ", topicName),
                            cause
                        );
                    }
                } catch (final InterruptedException interruptedException) {
                    throw new InterruptException(interruptedException);
                } finally {
                    futuresForTopic.remove(topicName);
                }
            }
        }
    }

    private void validatePartitionCount(final ValidationResult validationResult,
                                        final InternalTopicConfig topicConfig,
                                        final TopicDescription topicDescription) {
        final String topicName = topicConfig.name();
        final int requiredPartitionCount = topicConfig.numberOfPartitions()
            .orElseThrow(() -> new IllegalStateException("No partition count is specified for internal topic " +
                topicName + ". " + BUG_ERROR_MESSAGE));
        final int actualPartitionCount = topicDescription.partitions().size();
        if (actualPartitionCount != requiredPartitionCount) {
            validationResult.addMisconfiguration(
                topicName,
                "Internal topic " + topicName + " requires " + requiredPartitionCount + " partitions, " +
                "but the existing topic on the broker has " + actualPartitionCount + " partitions."
            );
        }
    }

    private void validateCleanupPolicy(final ValidationResult validationResult,
                                       final InternalTopicConfig topicConfig,
                                       final Config brokerSideTopicConfig) {
        if (topicConfig instanceof UnwindowedUnversionedChangelogTopicConfig) {
            validateCleanupPolicyForUnwindowedUnversionedChangelogs(validationResult, topicConfig, brokerSideTopicConfig);
        } else if (topicConfig instanceof WindowedChangelogTopicConfig) {
            validateCleanupPolicyForWindowedChangelogs(validationResult, topicConfig, brokerSideTopicConfig);
        } else if (topicConfig instanceof VersionedChangelogTopicConfig) {
            validateCleanupPolicyForVersionedChangelogs(validationResult, topicConfig, brokerSideTopicConfig);
        } else if (topicConfig instanceof RepartitionTopicConfig) {
            validateCleanupPolicyForRepartitionTopic(validationResult, topicConfig, brokerSideTopicConfig);
        } else {
            throw new IllegalStateException("Internal topic " + topicConfig.name() + " has unknown type.");
        }
    }

    private void validateCleanupPolicyForUnwindowedUnversionedChangelogs(final ValidationResult validationResult,
                                                                         final InternalTopicConfig topicConfig,
                                                                         final Config brokerSideTopicConfig) {
        final String topicName = topicConfig.name();
        final String cleanupPolicy = getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, topicName);
        if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
            validationResult.addMisconfiguration(
                topicName,
                "Cleanup policy (" + TopicConfig.CLEANUP_POLICY_CONFIG + ") of existing internal topic "
                    + topicName + " should not contain \""
                + TopicConfig.CLEANUP_POLICY_DELETE + "\"."
            );
        }
    }

    private void validateCleanupPolicyForWindowedChangelogs(final ValidationResult validationResult,
                                                            final InternalTopicConfig topicConfig,
                                                            final Config brokerSideTopicConfig) {
        final String topicName = topicConfig.name();
        final String cleanupPolicy = getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, topicName);
        if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
            final long brokerSideRetentionMs =
                Long.parseLong(getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_MS_CONFIG, topicName));
            final Map<String, String> streamsSideConfig =
                topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
            final long streamsSideRetentionMs = Long.parseLong(streamsSideConfig.get(TopicConfig.RETENTION_MS_CONFIG));
            if (brokerSideRetentionMs < streamsSideRetentionMs) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention time (" + TopicConfig.RETENTION_MS_CONFIG + ") of existing internal topic "
                        + topicName + " is " + brokerSideRetentionMs + " but should be " + streamsSideRetentionMs + " or larger."
                );
            }
            final String brokerSideRetentionBytes =
                getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_BYTES_CONFIG, topicName);
            if (brokerSideRetentionBytes != null) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention byte (" + TopicConfig.RETENTION_BYTES_CONFIG + ") of existing internal topic "
                        + topicName + " is set but it should be unset."
                );
            }
        }
    }

    private void validateCleanupPolicyForVersionedChangelogs(final ValidationResult validationResult,
                                                             final InternalTopicConfig topicConfig,
                                                             final Config brokerSideTopicConfig) {
        final String topicName = topicConfig.name();
        final String cleanupPolicy = getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, topicName);

        if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
            validationResult.addMisconfiguration(
                topicName,
                "Cleanup policy (" + TopicConfig.CLEANUP_POLICY_CONFIG + ") of existing internal topic "
                    + topicName + " should not contain \""
                    + TopicConfig.CLEANUP_POLICY_DELETE + "\"."
            );
        }

        final long brokerSideCompactionLagMs =
            Long.parseLong(getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, topicName));
        final Map<String, String> streamsSideConfig =
            topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
        final long streamsSideCompactionLagMs = Long.parseLong(streamsSideConfig.get(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG));
        if (brokerSideCompactionLagMs < streamsSideCompactionLagMs) {
            validationResult.addMisconfiguration(
                topicName,
                "Min compaction lag (" + TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG + ") of existing internal topic "
                    + topicName + " is " + brokerSideCompactionLagMs + " but should be " + streamsSideCompactionLagMs + " or larger."
            );
        }
    }

    private void validateCleanupPolicyForRepartitionTopic(final ValidationResult validationResult,
                                                          final InternalTopicConfig topicConfig,
                                                          final Config brokerSideTopicConfig) {
        final String topicName = topicConfig.name();
        final String cleanupPolicy = getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, topicName);
        if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_COMPACT)) {
            validationResult.addMisconfiguration(
                topicName,
                "Cleanup policy (" + TopicConfig.CLEANUP_POLICY_CONFIG + ") of existing internal topic "
                    + topicName + " should not contain \"" + TopicConfig.CLEANUP_POLICY_COMPACT + "\"."
            );
        } else if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
            final long brokerSideRetentionMs =
                Long.parseLong(getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_MS_CONFIG, topicName));
            if (brokerSideRetentionMs != -1) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention time (" + TopicConfig.RETENTION_MS_CONFIG + ") of existing internal topic "
                        + topicName + " is " + brokerSideRetentionMs + " but should be -1."
                );
            }
            final String brokerSideRetentionBytes =
                getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_BYTES_CONFIG, topicName);
            if (brokerSideRetentionBytes != null) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention byte (" + TopicConfig.RETENTION_BYTES_CONFIG + ") of existing internal topic "
                        + topicName + " is set but it should be unset."
                );
            }
        }
    }

    private String getBrokerSideConfigValue(final Config brokerSideTopicConfig,
                                            final String configName,
                                            final String topicName) {
        final ConfigEntry brokerSideConfigEntry = brokerSideTopicConfig.get(configName);
        if (brokerSideConfigEntry == null) {
            throw new IllegalStateException("The config " + configName + " for topic " +
                topicName + " could not be " + "retrieved from the brokers. " + BUG_ERROR_MESSAGE);
        }
        return brokerSideConfigEntry.value();
    }

    public Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics) {
        log.debug("Starting to describe topics {} in partition assignor.", topics);

        long currentWallClockMs = time.milliseconds();
        final long deadlineMs = currentWallClockMs + retryTimeoutMs;

        final Set<String> topicsToDescribe = new HashSet<>(topics);
        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new HashMap<>();

        while (!topicsToDescribe.isEmpty()) {
            final Map<String, List<TopicPartitionInfo>> existed = getTopicPartitionInfo(topicsToDescribe, null);
            topicPartitionInfo.putAll(existed);
            topicsToDescribe.removeAll(topicPartitionInfo.keySet());
            if (!topicsToDescribe.isEmpty()) {
                currentWallClockMs = time.milliseconds();

                if (currentWallClockMs >= deadlineMs) {
                    final String timeoutError = String.format(
                        "Could not create topics within %d milliseconds. " +
                            "This can happen if the Kafka cluster is temporarily not available.",
                        retryTimeoutMs);
                    log.error(timeoutError);
                    throw new TimeoutException(timeoutError);
                }
                log.info(
                    "Topics {} could not be describe fully. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                    topics,
                    retryBackOffMs,
                    deadlineMs - currentWallClockMs
                );
                Utils.sleep(retryBackOffMs);
            }
        }
        log.debug("Completed describing topics");
        return topicPartitionInfo;
    }

    /**
     * Prepares a set of given internal topics.
     *
     * If a topic does not exist creates a new topic.
     * If a topic with the correct number of partitions exists ignores it.
     * If a topic exists already but has different number of partitions we fail and throw exception requesting user to reset the app before restarting again.
     * @return the set of topics which had to be newly created
     */
    public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) {
        // we will do the validation / topic-creation in a loop, until we have confirmed all topics
        // have existed with the expected number of partitions, or some create topic returns fatal errors.
        log.debug("Starting to validate internal topics {} in partition assignor.", topics);

        long currentWallClockMs = time.milliseconds();
        final long deadlineMs = currentWallClockMs + retryTimeoutMs;

        Set<String> topicsNotReady = new HashSet<>(topics.keySet());
        final Set<String> newlyCreatedTopics = new HashSet<>();

        while (!topicsNotReady.isEmpty()) {
            final Set<String> tempUnknownTopics = new HashSet<>();
            topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
            newlyCreatedTopics.addAll(topicsNotReady);

            if (!topicsNotReady.isEmpty()) {
                final Set<NewTopic> newTopics = new HashSet<>();

                for (final String topicName : topicsNotReady) {
                    if (tempUnknownTopics.contains(topicName)) {
                        // for the tempUnknownTopics, don't create topic for them
                        // we'll check again later if remaining retries > 0
                        continue;
                    }
                    final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
                    final Map<String, String> topicConfig = internalTopicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);

                    log.debug("Going to create topic {} with {} partitions and config {}.",
                        internalTopicConfig.name(),
                        internalTopicConfig.numberOfPartitions(),
                        topicConfig);

                    newTopics.add(
                        new NewTopic(
                            internalTopicConfig.name(),
                            internalTopicConfig.numberOfPartitions(),
                            Optional.of(replicationFactor))
                            .configs(topicConfig));
                }

                // it's possible that although some topics are not ready yet because they
                // are temporarily not available, not that they do not exist; in this case
                // the new topics to create may be empty and hence we can skip here
                if (!newTopics.isEmpty()) {
                    final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);

                    for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
                        final String topicName = createTopicResult.getKey();
                        try {
                            createTopicResult.getValue().get();
                            topicsNotReady.remove(topicName);
                        } catch (final InterruptedException fatalException) {
                            // this should not happen; if it ever happens it indicate a bug
                            Thread.currentThread().interrupt();
                            log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
                            throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
                        } catch (final ExecutionException executionException) {
                            final Throwable cause = executionException.getCause();
                            if (cause instanceof TopicExistsException) {
                                // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
                                log.info(
                                        "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
                                                +
                                                "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
                                                +
                                                "Error message was: {}", topicName, retryBackOffMs,
                                        cause.toString());
                            } else {
                                log.error("Unexpected error during topic creation for {}.\n" +
                                        "Error message was: {}", topicName, cause.toString());

                                if (cause instanceof UnsupportedVersionException) {
                                    final String errorMessage = cause.getMessage();
                                    if (errorMessage != null &&
                                            errorMessage.startsWith("Creating topics with default partitions/replication factor are only supported in CreateTopicRequest version 4+")) {

                                        throw new StreamsException(String.format(
                                                "Could not create topic %s, because brokers don't support configuration replication.factor=-1."
                                                        + " You can change the replication.factor config or upgrade your brokers to version 2.4 or newer to avoid this error.",
                                                topicName)
                                        );
                                    }
                                } else if (cause instanceof TimeoutException) {
                                    log.error("Creating topic {} timed out.\n" +
                                            "Error message was: {}", topicName, cause.toString());
                                } else {
                                    throw new StreamsException(
                                            String.format("Could not create topic %s.", topicName),
                                            cause
                                    );
                                }
                            }
                        }
                    }
                }
            }

            if (!topicsNotReady.isEmpty()) {
                currentWallClockMs = time.milliseconds();

                if (currentWallClockMs >= deadlineMs) {
                    final String timeoutError = String.format("Could not create topics within %d milliseconds. " +
                        "This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs);
                    log.error(timeoutError);
                    throw new TimeoutException(timeoutError);
                }
                log.info(
                    "Topics {} could not be made ready. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                    topicsNotReady,
                    retryBackOffMs,
                    deadlineMs - currentWallClockMs
                );
                Utils.sleep(retryBackOffMs);
            }
        }
        log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);

        return newlyCreatedTopics;
    }

    /**
     * Try to get the partition information for the given topics; return the partition info for topics that already exists.
     *
     * Topics that were not able to get its description will simply not be returned
     */
    // visible for testing
    protected Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics,
                                                                          final Set<String> tempUnknownTopics) {
        final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topics);
        final Map<String, KafkaFuture<TopicDescription>> futures = describeTopicsResult.topicNameValues();

        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new HashMap<>();
        for (final Map.Entry<String, KafkaFuture<TopicDescription>> topicFuture : futures.entrySet()) {
            final String topicName = topicFuture.getKey();
            try {
                final TopicDescription topicDescription = topicFuture.getValue().get();
                topicPartitionInfo.put(topicName, topicDescription.partitions());
            } catch (final InterruptedException fatalException) {
                // this should not happen; if it ever happens it indicate a bug
                Thread.currentThread().interrupt();
                log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
                throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
            } catch (final ExecutionException couldNotDescribeTopicException) {
                final Throwable cause = couldNotDescribeTopicException.getCause();
                if (cause instanceof UnknownTopicOrPartitionException) {
                    // This topic didn't exist
                    log.debug("Topic {} is unknown or not found, hence not existed yet.\n" +
                        "Error message was: {}", topicName, cause.toString());
                } else if (cause instanceof LeaderNotAvailableException) {
                    if (tempUnknownTopics != null) {
                        tempUnknownTopics.add(topicName);
                    }
                    log.debug("The leader of topic {} is not available.\n" +
                        "Error message was: {}", topicName, cause.toString());
                } else if (cause instanceof TimeoutException) {
                    if (tempUnknownTopics != null) {
                        tempUnknownTopics.add(topicName);
                    }
                    log.debug("Describing topic {} (to get number of partitions) timed out.\n" +
                            "Error message was: {}", topicName, cause.toString());
                } else {
                    log.error("Unexpected error during topic description for {}.\n" +
                        "Error message was: {}", topicName, cause.toString());
                    throw new StreamsException(String.format("Could not create topic %s.", topicName), cause);
                }
            }
        }

        return topicPartitionInfo;
    }

    /**
     * Try to get the number of partitions for the given topics; return the number of partitions for topics that already exists.
     *
     * Topics that were not able to get its description will simply not be returned
     */
    // visible for testing
    protected Map<String, Integer> getNumPartitions(final Set<String> topics,
                                                    final Set<String> tempUnknownTopics) {
        log.debug("Trying to check if topics {} have been created with expected number of partitions.", topics);

        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = getTopicPartitionInfo(topics, tempUnknownTopics);
        return topicPartitionInfo.entrySet().stream().collect(Collectors.toMap(
            Entry::getKey, e -> e.getValue().size()));
    }

    /**
     * Check the existing topics to have correct number of partitions; and return the remaining topics that needs to be created
     */
    private Set<String> validateTopics(final Set<String> topicsToValidate,
                                       final Map<String, InternalTopicConfig> topicsMap,
                                       final Set<String> tempUnknownTopics) {
        if (!topicsMap.keySet().containsAll(topicsToValidate)) {
            throw new IllegalStateException("The topics map " + topicsMap.keySet() + " does not contain all the topics " +
                topicsToValidate + " trying to validate.");
        }

        final Map<String, Integer> existedTopicPartition = getNumPartitions(topicsToValidate, tempUnknownTopics);

        final Set<String> topicsToCreate = new HashSet<>();
        for (final String topicName : topicsToValidate) {
            final Optional<Integer> numberOfPartitions = topicsMap.get(topicName).numberOfPartitions();
            if (!numberOfPartitions.isPresent()) {
                log.error("Found undefined number of partitions for topic {}", topicName);
                throw new StreamsException("Topic " + topicName + " number of partitions not defined");
            }
            if (existedTopicPartition.containsKey(topicName)) {
                if (!existedTopicPartition.get(topicName).equals(numberOfPartitions.get())) {
                    final String errorMsg = String.format("Existing internal topic %s has invalid partitions: " +
                            "expected: %d; actual: %d. " +
                            "Use 'org.apache.kafka.tools.StreamsResetter' tool to clean up invalid topics before processing.",
                        topicName, numberOfPartitions.get(), existedTopicPartition.get(topicName));
                    log.error(errorMsg);
                    throw new StreamsException(errorMsg);
                }
            } else {
                topicsToCreate.add(topicName);
            }
        }

        return topicsToCreate;
    }

    /**
     * Sets up internal topics.
     *
     * Either the given topic are all created or the method fails with an exception.
     *
     * @param topicConfigs internal topics to setup
     */
    public void setup(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to setup internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final Map<String, Map<String, String>> streamsSideTopicConfigs = topicConfigs.values().stream()
            .collect(Collectors.toMap(
                InternalTopicConfig::name,
                topicConfig -> topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention)
            ));
        final Set<String> createdTopics = new HashSet<>();
        final Set<String> topicStillToCreate = new HashSet<>(topicConfigs.keySet());
        while (!topicStillToCreate.isEmpty()) {
            final Set<NewTopic> newTopics = topicStillToCreate.stream()
                .map(topicName -> new NewTopic(
                        topicName,
                        topicConfigs.get(topicName).numberOfPartitions(),
                        Optional.of(replicationFactor)
                    ).configs(streamsSideTopicConfigs.get(topicName))
                ).collect(Collectors.toSet());

            log.info("Going to create internal topics: " + newTopics);
            final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);

            processCreateTopicResults(createTopicsResult, topicStillToCreate, createdTopics, deadline);

            maybeSleep(Collections.singletonList(topicStillToCreate), deadline, "created");
        }

        log.info("Completed setup of internal topics {}.", topicConfigs.keySet());
    }

    private void processCreateTopicResults(final CreateTopicsResult createTopicsResult,
                                           final Set<String> topicStillToCreate,
                                           final Set<String> createdTopics,
                                           final long deadline) {
        final Map<String, Throwable> lastErrorsSeenForTopic = new HashMap<>();
        final Map<String, KafkaFuture<Void>> createResultForTopic = createTopicsResult.values();
        while (!createResultForTopic.isEmpty()) {
            for (final String topicName : new HashSet<>(topicStillToCreate)) {
                if (!createResultForTopic.containsKey(topicName)) {
                    cleanUpCreatedTopics(createdTopics);
                    throw new IllegalStateException("Create topic results do not contain internal topic " + topicName
                        + " to setup. " + BUG_ERROR_MESSAGE);
                }
                final KafkaFuture<Void> createResult = createResultForTopic.get(topicName);
                if (createResult.isDone()) {
                    try {
                        createResult.get();
                        createdTopics.add(topicName);
                        topicStillToCreate.remove(topicName);
                    } catch (final ExecutionException executionException) {
                        final Throwable cause = executionException.getCause();
                        if (cause instanceof TopicExistsException) {
                            lastErrorsSeenForTopic.put(topicName, cause);
                            log.info("Internal topic {} already exists. Topic is probably marked for deletion. " +
                                "Will retry to create this topic later (to let broker complete async delete operation first)",
                                topicName);
                        } else if (cause instanceof TimeoutException) {
                            lastErrorsSeenForTopic.put(topicName, cause);
                            log.info("Creating internal topic {} timed out.", topicName);
                        } else {
                            cleanUpCreatedTopics(createdTopics);
                            log.error("Unexpected error during creation of internal topic: ", cause);
                            throw new StreamsException(
                                String.format("Could not create internal topic %s for the following reason: ", topicName),
                                cause
                            );
                        }
                    } catch (final InterruptedException interruptedException) {
                        throw new InterruptException(interruptedException);
                    } finally {
                        createResultForTopic.remove(topicName);
                    }
                }
            }

            maybeThrowTimeoutExceptionDuringSetup(
                topicStillToCreate,
                createdTopics,
                lastErrorsSeenForTopic,
                deadline
            );

            if (!createResultForTopic.isEmpty()) {
                Utils.sleep(100);
            }
        }
    }

    private void cleanUpCreatedTopics(final Set<String> topicsToCleanUp) {
        log.info("Starting to clean up internal topics {}.", topicsToCleanUp);

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final Set<String> topicsStillToCleanup = new HashSet<>(topicsToCleanUp);
        while (!topicsStillToCleanup.isEmpty()) {
            log.info("Going to cleanup internal topics: " + topicsStillToCleanup);
            final DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsStillToCleanup);
            final Map<String, KafkaFuture<Void>> deleteResultForTopic = deleteTopicsResult.topicNameValues();
            while (!deleteResultForTopic.isEmpty()) {
                for (final String topicName : new HashSet<>(topicsStillToCleanup)) {
                    if (!deleteResultForTopic.containsKey(topicName)) {
                        throw new IllegalStateException("Delete topic results do not contain internal topic " + topicName
                            + " to clean up. " + BUG_ERROR_MESSAGE);
                    }
                    final KafkaFuture<Void> deleteResult = deleteResultForTopic.get(topicName);
                    if (deleteResult.isDone()) {
                        try {
                            deleteResult.get();
                            topicsStillToCleanup.remove(topicName);
                        } catch (final ExecutionException executionException) {
                            final Throwable cause = executionException.getCause();
                            if (cause instanceof UnknownTopicOrPartitionException) {
                                log.info("Internal topic {} to clean up is missing", topicName);
                            } else if (cause instanceof LeaderNotAvailableException) {
                                log.info("The leader of internal topic {} to clean up is not available.", topicName);
                            } else if (cause instanceof TimeoutException) {
                                log.info("Cleaning up internal topic {} timed out.", topicName);
                            } else {
                                log.error("Unexpected error during cleanup of internal topics: ", cause);
                                throw new StreamsException(
                                    String.format("Could not clean up internal topics %s, because during the cleanup " +
                                            "of topic %s the following error occurred: ",
                                        topicsStillToCleanup, topicName),
                                    cause
                                );
                            }
                        } catch (final InterruptedException interruptedException) {
                            throw new InterruptException(interruptedException);
                        } finally {
                            deleteResultForTopic.remove(topicName);
                        }
                    }
                }

                maybeThrowTimeoutException(
                    Collections.singletonList(topicsStillToCleanup),
                    deadline,
                    String.format("Could not cleanup internal topics within %d milliseconds. This can happen if the " +
                            "Kafka cluster is temporarily not available or the broker did not complete topic creation " +
                            "before the cleanup. The following internal topics could not be cleaned up: %s",
                        retryTimeoutMs, topicsStillToCleanup)
                );

                if (!deleteResultForTopic.isEmpty()) {
                    Utils.sleep(100);
                }
            }

            maybeSleep(
                Collections.singletonList(topicsStillToCleanup),
                deadline,
                "validated"
            );
        }

        log.info("Completed cleanup of internal topics {}.", topicsToCleanUp);
    }

    private void maybeThrowTimeoutException(final List<Set<String>> topicStillToProcess,
                                            final long deadline,
                                            final String errorMessage) {
        if (topicStillToProcess.stream().anyMatch(resultSet -> !resultSet.isEmpty())) {
            final long now = time.milliseconds();
            if (now >= deadline) {
                log.error(errorMessage);
                throw new TimeoutException(errorMessage);
            }
        }
    }

    private void maybeThrowTimeoutExceptionDuringSetup(final Set<String> topicStillToProcess,
                                                       final Set<String> createdTopics,
                                                       final Map<String, Throwable> lastErrorsSeenForTopic,
                                                       final long deadline) {
        if (topicStillToProcess.stream().anyMatch(resultSet -> !resultSet.isEmpty())) {
            final long now = time.milliseconds();
            if (now >= deadline) {
                cleanUpCreatedTopics(createdTopics);
                final String errorMessage = String.format("Could not create internal topics within %d milliseconds. This can happen if the " +
                    "Kafka cluster is temporarily not available or a topic is marked for deletion and the broker " +
                    "did not complete its deletion within the timeout. The last errors seen per topic are: %s",
                    retryTimeoutMs, lastErrorsSeenForTopic);
                log.error(errorMessage);
                throw new TimeoutException(errorMessage);
            }
        }
    }

    private void maybeSleep(final List<Set<String>> resultSetsStillToValidate,
                            final long deadline,
                            final String action) {
        if (resultSetsStillToValidate.stream().anyMatch(resultSet -> !resultSet.isEmpty())) {
            final long now = time.milliseconds();
            log.info(
                "Internal topics {} could not be {}. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                resultSetsStillToValidate.stream().flatMap(Collection::stream).collect(Collectors.toSet()),
                action,
                retryBackOffMs,
                deadline - now
            );
            Utils.sleep(retryBackOffMs);
        }
    }
}
