guozhangwang commented on a change in pull request #9848:
URL: https://github.com/apache/kafka/pull/9848#discussion_r559916595
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -365,7 +364,7 @@ public GroupAssignment assign(final Cluster metadata, final
GroupSubscription gr
// the maximum of the depending sub-topologies source topics'
number of partitions
final Map<Integer, TopicsInfo> topicGroups =
taskManager.builder().topicGroups();
- final Map<TopicPartition, PartitionInfo>
allRepartitionTopicPartitions = prepareRepartitionTopics(topicGroups, metadata);
+ final Map<TopicPartition, PartitionInfo>
allRepartitionTopicPartitions = prepareRepartitionTopics(metadata);
Review comment:
nit: line 365 above can be moved down now since it is only needed before
line 379.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionTopicsTest.java
##########
@@ -0,0 +1,429 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
Review comment:
As always, thanks for the great test coverage!
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.errors.MissingSourceTopicException;
+import org.apache.kafka.streams.errors.TaskAssignmentException;
+import
org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.TopicsInfo;
+import
org.apache.kafka.streams.processor.internals.assignment.CopartitionedTopicsEnforcer;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RepartitionTopics {
+
+ private final InternalTopicManager internalTopicManager;
+ private final InternalTopologyBuilder internalTopologyBuilder;
+ private final Cluster clusterMetadata;
+ private final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer;
+ private final Logger log;
+ private final Map<TopicPartition, PartitionInfo> topicPartitionInfos = new
HashMap<>();
+
+ public RepartitionTopics(final InternalTopologyBuilder
internalTopologyBuilder,
+ final InternalTopicManager internalTopicManager,
+ final CopartitionedTopicsEnforcer
copartitionedTopicsEnforcer,
+ final Cluster clusterMetadata,
+ final String logPrefix) {
+ this.internalTopologyBuilder = internalTopologyBuilder;
+ this.internalTopicManager = internalTopicManager;
+ this.clusterMetadata = clusterMetadata;
+ this.copartitionedTopicsEnforcer = copartitionedTopicsEnforcer;
+ final LogContext logContext = new LogContext(logPrefix);
+ log = logContext.logger(getClass());
+ }
+
+ public void setup() {
+ final Map<Integer, TopicsInfo> topicGroups =
internalTopologyBuilder.topicGroups();
+ final Map<String, InternalTopicConfig> repartitionTopicMetadata =
computeRepartitionTopicConfig(topicGroups, clusterMetadata);
+
+ // ensure the co-partitioning topics within the group have the same
number of partitions,
+ // and enforce the number of partitions for those repartition topics
to be the same if they
+ // are co-partitioned as well.
+ ensureCopartitioning(internalTopologyBuilder.copartitionGroups(),
repartitionTopicMetadata, clusterMetadata);
+
+ // make sure the repartition source topics exist with the right number
of partitions,
+ // create these topics if necessary
+ internalTopicManager.makeReady(repartitionTopicMetadata);
Review comment:
Not related to this PR: when reading the `makeReady` code again, I feel
that the way we get the retries config by first constructing a dummy admin
client can be simplified. Also, it is debatable that if we only do verify, then
do we really need to retry or not since the admin client itself already retries
internally.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
