<commit_before>/*
 * 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 org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.errors.MissingSourceTopicException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskAssignmentException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.TopicsInfo;
import org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
import org.apache.kafka.streams.processor.internals.assignment.CopartitionedTopicsEnforcer;

import java.util.LinkedList;
import java.util.Queue;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

import static org.apache.kafka.streams.processor.internals.TopologyMetadata.getTopologyNameOrElseUnnamed;

public class RepartitionTopics {

    private final InternalTopicManager internalTopicManager;
    private final TopologyMetadata topologyMetadata;
    private final Cluster clusterMetadata;
    private final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer;
    private final Logger log;
    private final Map<TopicPartition, PartitionInfo> topicPartitionInfos = new HashMap<>();
    private final Map<Subtopology, Set<String>> missingInputTopicsBySubtopology = new HashMap<>();

    public RepartitionTopics(final TopologyMetadata topologyMetadata,
                             final InternalTopicManager internalTopicManager,
                             final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer,
                             final Cluster clusterMetadata,
                             final String logPrefix) {
        this.topologyMetadata = topologyMetadata;
        this.internalTopicManager = internalTopicManager;
        this.clusterMetadata = clusterMetadata;
        this.copartitionedTopicsEnforcer = copartitionedTopicsEnforcer;
        final LogContext logContext = new LogContext(logPrefix);
        log = logContext.logger(getClass());
    }

    public void setup() {
        final Map<String, InternalTopicConfig> repartitionTopicMetadata = computeRepartitionTopicConfig(clusterMetadata);

        if (repartitionTopicMetadata.isEmpty()) {
            if (missingInputTopicsBySubtopology.isEmpty()) {
                log.info("Skipping the repartition topic validation since there are no repartition topics.");
            } else {
                log.info("Skipping the repartition topic validation since all topologies containing repartition"
                             + "topics are missing external user source topics and cannot be processed.");
            }
        } else {
            // ensure the co-partitioning topics within the group have the same number of partitions,
            // and enforce the number of partitions for those repartition topics to be the same if they
            // are co-partitioned as well.
            ensureCopartitioning(topologyMetadata.copartitionGroups(), repartitionTopicMetadata, clusterMetadata);

            // make sure the repartition source topics exist with the right number of partitions,
            // create these topics if necessary
            internalTopicManager.makeReady(repartitionTopicMetadata);

            // augment the metadata with the newly computed number of partitions for all the
            // repartition source topics
            for (final Map.Entry<String, InternalTopicConfig> entry : repartitionTopicMetadata.entrySet()) {
                final String topic = entry.getKey();
                final int numPartitions = entry.getValue().numberOfPartitions().orElse(-1);

                for (int partition = 0; partition < numPartitions; partition++) {
                    topicPartitionInfos.put(
                        new TopicPartition(topic, partition),
                        new PartitionInfo(topic, partition, null, new Node[0], new Node[0])
                    );
                }
            }
        }
    }

    public Set<String> topologiesWithMissingInputTopics() {
        return missingInputTopicsBySubtopology.keySet()
            .stream()
            .map(s -> getTopologyNameOrElseUnnamed(s.namedTopology))
            .collect(Collectors.toSet());
    }

    public Queue<StreamsException> missingSourceTopicExceptions() {
        return missingInputTopicsBySubtopology.entrySet().stream().map(entry -> {
            final Set<String> missingSourceTopics = entry.getValue();
            final int subtopologyId = entry.getKey().nodeGroupId;
            final String topologyName = entry.getKey().namedTopology;

            return new StreamsException(
                new MissingSourceTopicException(String.format(
                    "Missing source topics %s for subtopology %s of topology %s",
                    missingSourceTopics, subtopologyId, topologyName)),
                new TaskId(subtopologyId, 0, topologyName));
        }).collect(Collectors.toCollection(LinkedList::new));
    }

    public Map<TopicPartition, PartitionInfo> topicPartitionsInfo() {
        return Collections.unmodifiableMap(topicPartitionInfos);
    }

    /**
     * @param clusterMetadata  cluster metadata, eg which topics exist on the brokers
     */
    private Map<String, InternalTopicConfig> computeRepartitionTopicConfig(final Cluster clusterMetadata) {
        final Set<TopicsInfo> allTopicsInfo = new HashSet<>();
        final Map<String, InternalTopicConfig> allRepartitionTopicConfigs = new HashMap<>();
        for (final Map.Entry<String, Map<Subtopology, TopicsInfo>> topologyEntry : topologyMetadata.topologyToSubtopologyTopicsInfoMap().entrySet()) {
            final String topologyName = topologyMetadata.hasNamedTopologies() ? topologyEntry.getKey() : null;

            final Set<TopicsInfo> topicsInfoForTopology = new HashSet<>();
            final Set<String> missingSourceTopicsForTopology = new HashSet<>();
            final Map<String, InternalTopicConfig> repartitionTopicConfigsForTopology = new HashMap<>();

            for (final Map.Entry<Subtopology, TopicsInfo> subtopologyEntry : topologyEntry.getValue().entrySet()) {
                final TopicsInfo topicsInfo = subtopologyEntry.getValue();

                topicsInfoForTopology.add(topicsInfo);
                repartitionTopicConfigsForTopology.putAll(
                    topicsInfo.repartitionSourceTopics
                        .values()
                        .stream()
                        .collect(Collectors.toMap(InternalTopicConfig::name, topicConfig -> topicConfig)));

                final Set<String> missingSourceTopicsForSubtopology = computeMissingExternalSourceTopics(topicsInfo, clusterMetadata);
                missingSourceTopicsForTopology.addAll(missingSourceTopicsForSubtopology);
                if (!missingSourceTopicsForSubtopology.isEmpty()) {
                    missingInputTopicsBySubtopology.put(subtopologyEntry.getKey(), missingSourceTopicsForSubtopology);
                    log.error("Subtopology {} was missing source topics {} and will be excluded from the current assignment, "
                        + "this can be due to the consumer client's metadata being stale or because they have "
                        + "not been created yet. Please verify that you have created all input topics; if they "
                        + "do exist, you just need to wait for the metadata to be updated, at which time a new "
                        + "rebalance will be kicked off automatically and the topology will be retried at that time."
                        + topologyName, missingSourceTopicsForTopology);
                }
            }
            if (missingSourceTopicsForTopology.isEmpty()) {
                allRepartitionTopicConfigs.putAll(repartitionTopicConfigsForTopology);
                allTopicsInfo.addAll(topicsInfoForTopology);
            } else {
                log.debug("Skipping repartition topic validation for entire topology {} due to missing source topics {}",
                    topologyName, missingSourceTopicsForTopology);
            }
        }
        setRepartitionSourceTopicPartitionCount(allRepartitionTopicConfigs, allTopicsInfo, clusterMetadata);

        return allRepartitionTopicConfigs;
    }

    private void ensureCopartitioning(final Collection<Set<String>> copartitionGroups,
                                      final Map<String, InternalTopicConfig> repartitionTopicMetadata,
                                      final Cluster clusterMetadata) {
        for (final Set<String> copartitionGroup : copartitionGroups) {<commit_msg>MINOR: Fix log4j entry in RepartitionTopics (#11958)<commit_after>/*
 * 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 org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.errors.MissingSourceTopicException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskAssignmentException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.TopicsInfo;
import org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
import org.apache.kafka.streams.processor.internals.assignment.CopartitionedTopicsEnforcer;

import java.util.LinkedList;
import java.util.Queue;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

import static org.apache.kafka.streams.processor.internals.TopologyMetadata.getTopologyNameOrElseUnnamed;

public class RepartitionTopics {

    private final InternalTopicManager internalTopicManager;
    private final TopologyMetadata topologyMetadata;
    private final Cluster clusterMetadata;
    private final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer;
    private final Logger log;
    private final Map<TopicPartition, PartitionInfo> topicPartitionInfos = new HashMap<>();
    private final Map<Subtopology, Set<String>> missingInputTopicsBySubtopology = new HashMap<>();

    public RepartitionTopics(final TopologyMetadata topologyMetadata,
                             final InternalTopicManager internalTopicManager,
                             final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer,
                             final Cluster clusterMetadata,
                             final String logPrefix) {
        this.topologyMetadata = topologyMetadata;
        this.internalTopicManager = internalTopicManager;
        this.clusterMetadata = clusterMetadata;
        this.copartitionedTopicsEnforcer = copartitionedTopicsEnforcer;
        final LogContext logContext = new LogContext(logPrefix);
        log = logContext.logger(getClass());
    }

    public void setup() {
        final Map<String, InternalTopicConfig> repartitionTopicMetadata = computeRepartitionTopicConfig(clusterMetadata);

        if (repartitionTopicMetadata.isEmpty()) {
            if (missingInputTopicsBySubtopology.isEmpty()) {
                log.info("Skipping the repartition topic validation since there are no repartition topics.");
            } else {
                log.info("Skipping the repartition topic validation since all topologies containing repartition"
                             + "topics are missing external user source topics and cannot be processed.");
            }
        } else {
            // ensure the co-partitioning topics within the group have the same number of partitions,
            // and enforce the number of partitions for those repartition topics to be the same if they
            // are co-partitioned as well.
            ensureCopartitioning(topologyMetadata.copartitionGroups(), repartitionTopicMetadata, clusterMetadata);

            // make sure the repartition source topics exist with the right number of partitions,
            // create these topics if necessary
            internalTopicManager.makeReady(repartitionTopicMetadata);

            // augment the metadata with the newly computed number of partitions for all the
            // repartition source topics
            for (final Map.Entry<String, InternalTopicConfig> entry : repartitionTopicMetadata.entrySet()) {
                final String topic = entry.getKey();
                final int numPartitions = entry.getValue().numberOfPartitions().orElse(-1);

                for (int partition = 0; partition < numPartitions; partition++) {
                    topicPartitionInfos.put(
                        new TopicPartition(topic, partition),
                        new PartitionInfo(topic, partition, null, new Node[0], new Node[0])
                    );
                }
            }
        }
    }

    public Set<String> topologiesWithMissingInputTopics() {
        return missingInputTopicsBySubtopology.keySet()
            .stream()
            .map(s -> getTopologyNameOrElseUnnamed(s.namedTopology))
            .collect(Collectors.toSet());
    }

    public Queue<StreamsException> missingSourceTopicExceptions() {
        return missingInputTopicsBySubtopology.entrySet().stream().map(entry -> {
            final Set<String> missingSourceTopics = entry.getValue();
            final int subtopologyId = entry.getKey().nodeGroupId;
            final String topologyName = entry.getKey().namedTopology;

            return new StreamsException(
                new MissingSourceTopicException(String.format(
                    "Missing source topics %s for subtopology %s of topology %s",
                    missingSourceTopics, subtopologyId, topologyName)),
                new TaskId(subtopologyId, 0, topologyName));
        }).collect(Collectors.toCollection(LinkedList::new));
    }

    public Map<TopicPartition, PartitionInfo> topicPartitionsInfo() {
        return Collections.unmodifiableMap(topicPartitionInfos);
    }

    /**
     * @param clusterMetadata  cluster metadata, eg which topics exist on the brokers
     */
    private Map<String, InternalTopicConfig> computeRepartitionTopicConfig(final Cluster clusterMetadata) {
        final Set<TopicsInfo> allTopicsInfo = new HashSet<>();
        final Map<String, InternalTopicConfig> allRepartitionTopicConfigs = new HashMap<>();
        for (final Map.Entry<String, Map<Subtopology, TopicsInfo>> topologyEntry : topologyMetadata.topologyToSubtopologyTopicsInfoMap().entrySet()) {
            final String topologyName = topologyMetadata.hasNamedTopologies() ? topologyEntry.getKey() : null;

            final Set<TopicsInfo> topicsInfoForTopology = new HashSet<>();
            final Set<String> missingSourceTopicsForTopology = new HashSet<>();
            final Map<String, InternalTopicConfig> repartitionTopicConfigsForTopology = new HashMap<>();

            for (final Map.Entry<Subtopology, TopicsInfo> subtopologyEntry : topologyEntry.getValue().entrySet()) {
                final TopicsInfo topicsInfo = subtopologyEntry.getValue();

                topicsInfoForTopology.add(topicsInfo);
                repartitionTopicConfigsForTopology.putAll(
                    topicsInfo.repartitionSourceTopics
                        .values()
                        .stream()
                        .collect(Collectors.toMap(InternalTopicConfig::name, topicConfig -> topicConfig)));

                final Set<String> missingSourceTopicsForSubtopology = computeMissingExternalSourceTopics(topicsInfo, clusterMetadata);
                missingSourceTopicsForTopology.addAll(missingSourceTopicsForSubtopology);
                if (!missingSourceTopicsForSubtopology.isEmpty()) {
                    final Subtopology subtopology = subtopologyEntry.getKey();
                    missingInputTopicsBySubtopology.put(subtopology, missingSourceTopicsForSubtopology);
                    log.error("Subtopology {} was missing source topics {} and will be excluded from the current assignment, "
                        + "this can be due to the consumer client's metadata being stale or because they have "
                        + "not been created yet. Please verify that you have created all input topics; if they "
                        + "do exist, you just need to wait for the metadata to be updated, at which time a new "
                        + "rebalance will be kicked off automatically and the topology will be retried at that time.",
                        subtopology.nodeGroupId, missingSourceTopicsForSubtopology);
                }
            }

            if (missingSourceTopicsForTopology.isEmpty()) {
                allRepartitionTopicConfigs.putAll(repartitionTopicConfigsForTopology);
                allTopicsInfo.addAll(topicsInfoForTopology);
            } else {
                log.debug("Skipping repartition topic validation for entire topology {} due to missing source topics {}",
                    topologyName, missingSourceTopicsForTopology);
            }
        }
        setRepartitionSourceTopicPartitionCount(allRepartitionTopicConfigs, allTopicsInfo, clusterMetadata);

        return allRepartitionTopicConfigs;
    }

    private void ensureCopartitioning(final Collection<Set<String>> copartitionGroups,
                                      final Map<String, InternalTopicConfig> repartitionTopicMetadata,
                                      final Cluster clusterMetadata) {
        for (final Set<String> copartitionGroup : copartitionGroups) {
{'fraction_non_alphanumeric': 0.07262825732899023, 'fraction_numerical': 0.0013232899022801302, 'mean_word_length': 2.865630533149715, 'pattern_counts': {'":': 0, '<': 65, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6965609', 'n_tokens_mistral': 4834, 'n_tokens_neox': 4673, 'n_words': 1294}