lucasbru commented on code in PR #18397: URL: https://github.com/apache/kafka/pull/18397#discussion_r1908706477
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.errors.StreamsInvalidTopologyException; +import org.apache.kafka.common.utils.LogContext; + +import org.slf4j.Logger; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.OptionalInt; +import java.util.Set; +import java.util.TreeMap; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * This class is responsible for enforcing the number of partitions in copartitioned topics. For each copartition group, it checks whether + * the number of partitions for all repartition topics is the same, and enforces copartitioning for repartition topics whose number of + * partitions is not enforced by the topology. + */ +public class CopartitionedTopicsEnforcer { + + private final Logger log; + private final Function<String, OptionalInt> topicPartitionCountProvider; + + /** + * The constructor for the class. + * + * @param logContext The context for emitting log messages. + * @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the broker + * as well as any partition number decisions that have already been made. In particular, we expect + * the number of partitions for all repartition topics defined, even if they do not exist in the + * broker yet. + */ + public CopartitionedTopicsEnforcer(final LogContext logContext, + final Function<String, OptionalInt> topicPartitionCountProvider) { + this.log = logContext.logger(getClass()); + this.topicPartitionCountProvider = topicPartitionCountProvider; + } + + /** + * Enforces the number of partitions for copartitioned topics. + * + * @param copartitionedTopics The set of copartitioned topics (external source topics and repartition topics). + * @param fixedRepartitionTopics The set of repartition topics whose partition count is fixed by the topology. Review Comment: I don't want too restrictive - the number of partitions is fixed in the topology, and on the broker we don't necessarily know why it was fixed. But I think right now you are right, this only happens when the user explicitly specifies it in `repartition`, so I updated it this way: ``` The set of repartition topics whose partition count is fixed by the topology sent by the client (in particular, when the user uses `repartition` in the DSL). ``` -- 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]
