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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
