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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
