[PR] KAFKA-18311: Internal Topic Manager [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru opened a new pull request, #18442:
URL: https://github.com/apache/kafka/pull/18442

   The internal topic manager takes a requested topology and returns a configured topology, as well as any internal topics that need to be created.
   
   Shares with the client-side "InternalTopicManager" the name only.
   
   This is a stacked PR. Only review the last commit
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru commented on PR #18442:
URL: https://github.com/apache/kafka/pull/18442#issuecomment-2582922664

   @cadonna I moved the source topic check as you suggested.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922205600


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   On the following page it says:
   ```
   If a method call is longer than 120 characters, switch to a single parameter per line formatting (instead of just breaking it into two lines only).
   ```
   https://kafka.apache.org/coding-guide
   
   In the past, we accepted both
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology
   ) {
   ```
   or  
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology) {
   ```
   
   In this specific case -- I do not know why -- I thought one line might be enough.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922564511


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -148,6 +148,19 @@ private static Map<String, Integer> decidePartitionCounts(final LogContext logCo
 
         decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
 
+        enforceCopartitioning(topology, copartitionGroupsBySubtopology, log,
+            decidedPartitionCountsForInternalTopics, copartitionedTopicsEnforcer);

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922207534


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -148,6 +148,19 @@ private static Map<String, Integer> decidePartitionCounts(final LogContext logCo
 
         decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
 
+        enforceCopartitioning(topology, copartitionGroupsBySubtopology, log,
+            decidedPartitionCountsForInternalTopics, copartitionedTopicsEnforcer);

Review Comment:
   nit:
   ```suggestion
           enforceCopartitioning(
               topology, 
               copartitionGroupsBySubtopology, 
               log,
               decidedPartitionCountsForInternalTopics, 
               copartitionedTopicsEnforcer
           );
   ```
   documented here: https://kafka.apache.org/coding-guide
   There are always exceptions like log-messages.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   I agree but I do also not know where it comes from. I would put it in one line although it exceeds the 120 characters, but it is also fine as you did it. The best would be to have an automatic formatter, but we never manged in the past to setup one. There was an attempt: https://github.com/apache/kafka/pull/10428 but it was never merged. 
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922100063


##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopicMetadataTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {
+        Exception exception = assertThrows(IllegalArgumentException.class, () ->
+            new TopicMetadata(Uuid.ZERO_UUID, "valid-topic", 3, new HashMap<>()));
+        assertEquals("Topic id cannot be ZERO_UUID.", exception.getMessage());
+    }
+
+    @Test
+    public void testConstructorWithNullUuid() {
+        assertThrows(NullPointerException.class, () ->
+            new TopicMetadata(null, "valid-topic", 3, new HashMap<>()));
+    }
+
+    @Test
+    public void testConstructorWithNullName() {
+        assertThrows(NullPointerException.class, () ->
+            new TopicMetadata(Uuid.randomUuid(), null, 3, new HashMap<>()));
+    }
+
+    @Test
+    public void testConstructorWithEmptyName() {
+        Exception exception = assertThrows(IllegalArgumentException.class, () ->
+            new TopicMetadata(Uuid.randomUuid(), "", 3, new HashMap<>()));
+        assertEquals("Topic name cannot be empty.", exception.getMessage());
+    }
+
+    @Test
+    public void testConstructorWithNegativeNumPartitions() {

Review Comment:
   Hmm, actually not sure why the implementation allows this. I changed it to require positive, let's see if it breaks anything.



##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopicMetadataTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   This will make the line too long, and my IDE (based on imported checkstyle config) will wrap it the same way.... I added another line break.
   
   I didn't know this was an accepted standard in the streams code that everybody follows. Is there something I can do in my IDE to follow this? I imported the official Kafka checkstyle configuration, and it produces this kind of wrapping. It would be good if our "review nitpicking" and automated formatting tools somewhat agree.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.
+ * @param name           The topic name.
+ * @param numPartitions  The number of partitions.
+ * @param partitionRacks Map of every partition ID to a set of its rack IDs, if they exist. If rack information is unavailable for all
+ *                       partitions, this is an empty map.
+ */
+public record TopicMetadata(Uuid id, String name, int numPartitions, Map<Integer, Set<String>> partitionRacks) {
+
+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {
+        this.id = Objects.requireNonNull(id);
+        if (Uuid.ZERO_UUID.equals(id)) {
+            throw new IllegalArgumentException("Topic id cannot be ZERO_UUID.");
+        }
+        this.name = Objects.requireNonNull(name);
+        if (name.isEmpty()) {
+            throw new IllegalArgumentException("Topic name cannot be empty.");
+        }
+        this.numPartitions = numPartitions;
+        if (numPartitions < 0) {
+            throw new IllegalArgumentException("Number of partitions cannot be negative.");
+        }
+        this.partitionRacks = Objects.requireNonNull(partitionRacks);
+    }
+
+    public static TopicMetadata fromRecord(
+        StreamsGroupPartitionMetadataValue.TopicMetadata record
+    ) {

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.
+ * @param name           The topic name.
+ * @param numPartitions  The number of partitions.
+ * @param partitionRacks Map of every partition ID to a set of its rack IDs, if they exist. If rack information is unavailable for all
+ *                       partitions, this is an empty map.
+ */
+public record TopicMetadata(Uuid id, String name, int numPartitions, Map<Integer, Set<String>> partitionRacks) {
+
+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {

Review Comment:
   Done



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922205600


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   On the following page it says:
   ```
   If a method call is longer than 120 characters, switch to a single parameter per line formatting (instead of just breaking it into two lines only).
   ```
   https://kafka.apache.org/coding-guide
   
   In the past, we accepted both
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology
   ) {
   ```
   and 
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology) {
   ```
   
   In this specific case -- I do not know why -- I thought one line might be enough.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru commented on PR #18442:
URL: https://github.com/apache/kafka/pull/18442#issuecomment-2577589618

   PTAL @aliehsaeedii 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1922205600


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   On the following page it says:
   ```
   If a method call is longer than 120 characters, switch to a single parameter per line formatting (instead of just breaking it into two lines only).
   ```
   https://kafka.apache.org/coding-guide
   
   In the past, we accepted both
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology
   ) {
   ```
   and 
   ```java
   private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
       final StreamsGroupTopologyValue.Subtopology subtopology) {
   ```
   
   In this specific case -- I do not know why -- I thought one line might be enough.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru merged PR #18442:
URL: https://github.com/apache/kafka/pull/18442


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-18311: Internal Topic Manager (5/5) [kafka]

["cadonna (via GitHub)" <gi...@apache.org>]

cadonna commented on code in PR #18442:
URL: https://github.com/apache/kafka/pull/18442#discussion_r1912862606


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.

Review Comment:
   nit:
   ```suggestion
    * @param id             The topic ID.
   ```



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.
+ * @param name           The topic name.
+ * @param numPartitions  The number of partitions.
+ * @param partitionRacks Map of every partition ID to a set of its rack IDs, if they exist. If rack information is unavailable for all
+ *                       partitions, this is an empty map.
+ */
+public record TopicMetadata(Uuid id, String name, int numPartitions, Map<Integer, Set<String>> partitionRacks) {
+
+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {
+        this.id = Objects.requireNonNull(id);
+        if (Uuid.ZERO_UUID.equals(id)) {
+            throw new IllegalArgumentException("Topic id cannot be ZERO_UUID.");
+        }
+        this.name = Objects.requireNonNull(name);
+        if (name.isEmpty()) {
+            throw new IllegalArgumentException("Topic name cannot be empty.");
+        }
+        this.numPartitions = numPartitions;
+        if (numPartitions < 0) {
+            throw new IllegalArgumentException("Number of partitions cannot be negative.");
+        }
+        this.partitionRacks = Objects.requireNonNull(partitionRacks);
+    }
+
+    public static TopicMetadata fromRecord(
+        StreamsGroupPartitionMetadataValue.TopicMetadata record
+    ) {

Review Comment:
   ```suggestion
       public static TopicMetadata fromRecord(StreamsGroupPartitionMetadataValue.TopicMetadata record) {
   ```



##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopicMetadataTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {
+        Exception exception = assertThrows(IllegalArgumentException.class, () ->
+            new TopicMetadata(Uuid.ZERO_UUID, "valid-topic", 3, new HashMap<>()));
+        assertEquals("Topic id cannot be ZERO_UUID.", exception.getMessage());
+    }
+
+    @Test
+    public void testConstructorWithNullUuid() {
+        assertThrows(NullPointerException.class, () ->
+            new TopicMetadata(null, "valid-topic", 3, new HashMap<>()));
+    }
+
+    @Test
+    public void testConstructorWithNullName() {
+        assertThrows(NullPointerException.class, () ->
+            new TopicMetadata(Uuid.randomUuid(), null, 3, new HashMap<>()));
+    }
+
+    @Test
+    public void testConstructorWithEmptyName() {
+        Exception exception = assertThrows(IllegalArgumentException.class, () ->
+            new TopicMetadata(Uuid.randomUuid(), "", 3, new HashMap<>()));
+        assertEquals("Topic name cannot be empty.", exception.getMessage());
+    }
+
+    @Test
+    public void testConstructorWithNegativeNumPartitions() {

Review Comment:
   Could you add a test with partition number 0?
   This seems a special case that should be documented with a unit test.



##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopicMetadataTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {

Review Comment:
   A positive constructor unit test is missing.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
+            throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
+        }
+
+        return new ConfiguredInternalTopic(
+            topicInfo.name(),
+            topicInfo.partitions() == 0 ? decidedPartitionCountsForInternalTopics.get(topicInfo.name()) : topicInfo.partitions(),
+            topicInfo.replicationFactor() == 0 ? Optional.empty()
+                : Optional.of(topicInfo.replicationFactor()),
+            topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
+                .collect(Collectors.toMap(StreamsGroupTopologyValue.TopicConfig::key,
+                    StreamsGroupTopologyValue.TopicConfig::value))
+                : Collections.emptyMap()
+        );
+    }
+
+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {

Review Comment:
   ```suggestion
       private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(final StreamsGroupTopologyValue.Subtopology subtopology) {
   ```



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }

Review Comment:
   Maybe export this block into a private method, so that we have each of repartition topics setup, co-partition enforcer, and changelog topcis setup in one line. Just a proposal. 



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {

Review Comment:
   ```suggestion
       private static ConfiguredSubtopology fromPersistedSubtopology(final StreamsGroupTopologyValue.Subtopology subtopology, 
                                                                     final Map<String, Integer> decidedPartitionCountsForInternalTopics
       ) {
   ```



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.
+ * @param name           The topic name.
+ * @param numPartitions  The number of partitions.
+ * @param partitionRacks Map of every partition ID to a set of its rack IDs, if they exist. If rack information is unavailable for all
+ *                       partitions, this is an empty map.
+ */
+public record TopicMetadata(Uuid id, String name, int numPartitions, Map<Integer, Set<String>> partitionRacks) {
+
+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {

Review Comment:
   The following is the code style we usually use in Streams and that is more readable IMO. 
   ```suggestion
       public TopicMetadata(Uuid id,
                            String name,
                            int numPartitions,
                            Map<Integer, Set<String>> partitionRacks) {
   ```



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.coordinator.group.streams.topics;
+
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
+
+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.OptionalInt;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Responsible for configuring internal topics for a given topology.
+ */
+public class InternalTopicManager {
+
+    /**
+     * Configures the internal topics for the given topology. Given a topology and the topic metadata, this method determines the number of
+     * partitions for all internal topics and returns a {@link ConfiguredTopology} object.
+     *
+     * @param logContext    The log context.
+     * @param topology      The topology.
+     * @param topicMetadata The topic metadata.
+     * @return The configured topology.
+     */
+    public static ConfiguredTopology configureTopics(LogContext logContext,
+                                                     StreamsTopology topology,
+                                                     Map<String, TopicMetadata> topicMetadata) {
+        final Logger log = logContext.logger(InternalTopicManager.class);
+        final Collection<StreamsGroupTopologyValue.Subtopology> subtopologies = topology.subtopologies().values();
+
+        final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
+            subtopologies.stream()
+                .collect(Collectors.toMap(
+                    StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                    InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
+                );
+
+        try {
+            Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+            throwOnMissingSourceTopics(topology, topicMetadata);
+
+            Map<String, Integer> decidedPartitionCountsForInternalTopics =
+                decidePartitionCounts(logContext, topology, topicMetadata, copartitionGroupsBySubtopology, log);
+
+            final Map<String, ConfiguredSubtopology> configuredSubtopologies =
+                subtopologies.stream()
+                    .collect(Collectors.toMap(
+                        StreamsGroupTopologyValue.Subtopology::subtopologyId,
+                        x -> fromPersistedSubtopology(x, decidedPartitionCountsForInternalTopics))
+                    );
+
+            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topicMetadata);
+            if (!internalTopicsToCreate.isEmpty()) {
+                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
+                    "Internal topics are missing: " + internalTopicsToCreate.keySet()
+                ));
+                log.info("Valid topic configuration found, but internal topics are missing for topology epoch {}: {}",
+                    topology.topologyEpoch(), topicConfigurationException.get().toString());
+            } else {
+                log.info("Valid topic configuration found, topology epoch {} is now initialized.", topology.topologyEpoch());
+            }
+
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.of(configuredSubtopologies),
+                internalTopicsToCreate,
+                topicConfigurationException
+            );
+
+        } catch (TopicConfigurationException e) {
+            log.warn("Topic configuration failed for topology epoch {}: {} ",
+                topology.topologyEpoch(), e.toString());
+            return new ConfiguredTopology(
+                topology.topologyEpoch(),
+                Optional.empty(),
+                Map.of(),
+                Optional.of(e)
+            );
+        }
+    }
+
+    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
+                                                   final Map<String, TopicMetadata> topicMetadata) {
+        TreeSet<String> sortedMissingTopics = new TreeSet<>();
+        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
+            for (String sourceTopic : subtopology.sourceTopics()) {
+                if (!topicMetadata.containsKey(sourceTopic)) {
+                    sortedMissingTopics.add(sourceTopic);
+                }
+            }
+        }
+        if (!sortedMissingTopics.isEmpty()) {
+            throw TopicConfigurationException.missingSourceTopics(
+                "Source topics " + String.join(", ", sortedMissingTopics) + " are missing.");
+        }
+    }
+
+    private static Map<String, Integer> decidePartitionCounts(final LogContext logContext,
+                                                              final StreamsTopology topology,
+                                                              final Map<String, TopicMetadata> topicMetadata,
+                                                              final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
+                                                              final Logger log) {
+        final Map<String, Integer> decidedPartitionCountsForInternalTopics = new HashMap<>();
+        final Function<String, OptionalInt> topicPartitionCountProvider =
+            topic -> getPartitionCount(topicMetadata, topic, decidedPartitionCountsForInternalTopics);
+        final RepartitionTopics repartitionTopics = new RepartitionTopics(
+            logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+        final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(
+            logContext,
+            topicPartitionCountProvider);
+        final ChangelogTopics changelogTopics = new ChangelogTopics(logContext,
+            topology.subtopologies().values(),
+            topicPartitionCountProvider);
+
+        decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());
+
+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } 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.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }
+
+        decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());
+
+        return decidedPartitionCountsForInternalTopics;
+    }
+
+    private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
+                                                                     Map<String, TopicMetadata> topicMetadata) {
+
+        final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
+        for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
+            subtopology.repartitionSourceTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+            subtopology.stateChangelogTopics().values()
+                .forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
+        }
+        for (Map.Entry<String, TopicMetadata> topic : topicMetadata.entrySet()) {
+            final TopicMetadata existingTopic = topic.getValue();
+            final CreatableTopic expectedTopic = topicsToCreate.remove(topic.getKey());
+            if (expectedTopic != null) {
+                if (existingTopic.numPartitions() != expectedTopic.numPartitions()) {
+                    throw TopicConfigurationException.incorrectlyPartitionedTopics("Existing topic " + topic.getKey() + " has different"
+                        + " number of partitions: expected " + expectedTopic.numPartitions() + ", found " + existingTopic.numPartitions());
+                }
+            }
+        }
+        return topicsToCreate;
+    }
+
+    private static OptionalInt getPartitionCount(Map<String, TopicMetadata> topicMetadata,
+                                                 String topic,
+                                                 Map<String, Integer> decidedPartitionCountsForInternalTopics) {
+        final TopicMetadata metadata = topicMetadata.get(topic);
+        if (metadata == null) {
+            if (decidedPartitionCountsForInternalTopics.containsKey(topic)) {
+                return OptionalInt.of(decidedPartitionCountsForInternalTopics.get(topic));
+            } else {
+                return OptionalInt.empty();
+            }
+        } else {
+            return OptionalInt.of(metadata.numPartitions());
+        }
+    }
+
+    private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic config) {
+
+        final CreatableTopic creatableTopic = new CreatableTopic();
+
+        creatableTopic.setName(config.name());
+        creatableTopic.setNumPartitions(config.numberOfPartitions());
+
+        if (config.replicationFactor().isPresent() && config.replicationFactor().get() != 0) {
+            creatableTopic.setReplicationFactor(config.replicationFactor().get());
+        } else {
+            creatableTopic.setReplicationFactor((short) -1);
+        }
+
+        final CreatableTopicConfigCollection topicConfigs = new CreatableTopicConfigCollection();
+
+        config.topicConfigs().forEach((k, v) -> {
+            final CreatableTopicConfig topicConfig = new CreatableTopicConfig();
+            topicConfig.setName(k);
+            topicConfig.setValue(v);
+            topicConfigs.add(topicConfig);
+        });
+
+        creatableTopic.setConfigs(topicConfigs);
+
+        return creatableTopic;
+    }
+
+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {
+        return new ConfiguredSubtopology(
+            new HashSet<>(subtopology.sourceTopics()),
+            subtopology.repartitionSourceTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x)),
+            new HashSet<>(subtopology.repartitionSinkTopics()),
+            subtopology.stateChangelogTopics().stream()
+                .map(x -> fromPersistedTopicInfo(x, decidedPartitionCountsForInternalTopics))
+                .collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x))
+        );
+    }
+
+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {

Review Comment:
   ```suggestion
       private static ConfiguredInternalTopic fromPersistedTopicInfo(final StreamsGroupTopologyValue.TopicInfo topicInfo,
                                                                     final Map<String, Integer> decidedPartitionCountsForInternalTopics) {
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org