You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apourchet (via GitHub)" <gi...@apache.org> on 2024/05/15 21:58:35 UTC
[PR] KAFKA-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
apourchet opened a new pull request, #15972:
URL: https://github.com/apache/kafka/pull/15972
This rack information is required to compute rack-aware assignments, which many of the current assigners do.
The internal ClientMetadata class was also edited to pass around this rack information.
### 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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1605659907
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
+ sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
+ changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
}
+ final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
+ partitionGrouper.partitionGroups(sourceTopicsByGroup, cluster);
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask =
+ partitionGrouper.partitionGroups(changelogTopicsByGroup, cluster);
+
+ final Set<TaskId> logicalTaskIds = new HashSet<>();
+ final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
+ sourcePartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ sourceTopicPartitions.addAll(partitions);
+ });
+ final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
+ changelogPartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
Review Comment:
To be more precise, I'm imagining something like this:
```
final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
final Set<TopicPartition> nonSourceChangelogTopicPartitions = new HashSet<>();
for (final var entry : sourceTopicPartitions.entrySet()) {
final TaskId task = entry.getKey();
final Set<TopicPartition> taskSourcePartitions = entry.getValue();
final Set<TopicPartition> taskChangelogPartitions = changelogTopicPartitions.get(taskId);
final Set<TopicPartition> taskNonSourceChangelogPartitions = new HashSet(taskChangelogPartitions);
taskNonSourceChangelogPartitions.removeAll(taskSourcePartitions);
logicalTaskIds.add(taskId);
sourceTopicPartitions.addAll(taskSourcePartitions);
changelogTopicPartitions.addAll(taskChangelogPartitions);
nonSourceChangelogTopicPartitions.addAll(taskNonSourceChangelogPartitions);
}
```
Then we pass the `nonSourceChangelogPartitions` into the `#getRacksForTopicPartition` instead of the `changelogPartitions` set.
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15972:
URL: https://github.com/apache/kafka/pull/15972#issuecomment-2126312436
Hey @ableegoldman @apourchet, I see new failures in trunk that seems to be related to this PR. The last build of this PR had 100+ failures: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-15972/8/tests. Could you please take a look?
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on PR #15972:
URL: https://github.com/apache/kafka/pull/15972#issuecomment-2127918475
Yep, just noticed this. Sorry about that. We're taking a look
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "apourchet (via GitHub)" <gi...@apache.org>.
apourchet commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1606913055
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
Review Comment:
I agree, I wrote it this way to mimic the exact pattern of use that the RackAwareAssigner uses. Once this is all wired and tested though we can make optimizations and changes like this one (and the lazy rack info one).
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1605639072
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
+ sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
+ changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
}
+ final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
+ partitionGrouper.partitionGroups(sourceTopicsByGroup, cluster);
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask =
+ partitionGrouper.partitionGroups(changelogTopicsByGroup, cluster);
+
+ final Set<TaskId> logicalTaskIds = new HashSet<>();
+ final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
+ sourcePartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ sourceTopicPartitions.addAll(partitions);
+ });
+ final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
+ changelogPartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
Review Comment:
I suppose this doesn't hurt anything since `logicalTasks` is a Set, but the taskIds returned by the partition grouper should be the same for the source and changelog topics. So you can remove this line
(alternatively you can create the `logicalTaskIds` map up front by copying the keyset of one of the partitionsForTask maps but that's just an implementation detail, up to you. However I would probably consider adding a check to make sure these two maps return the same set of tasks. Doesn't need to scan the entire thing, maybe just a simple
```
if (sourcePartitionsForTask.size() != changelogPartitionsForTask.size()) {
log.error("Partition grouper returned {} tasks for source topics but {} tasks for changelog topics,
sourcePartitionsForTask.size(), changelogPartitionsForTask.size());
throw new TaskAssignmentException(//error msg );
}
##########
streams/src/main/java/org/apache/kafka/streams/processor/assignment/TaskInfo.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.assignment;
+
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.processor.TaskId;
+
+/**
+ * A simple container class corresponding to a given {@link TaskId}.
+ * Includes metadata such as whether it's stateful and the names of all state stores
+ * belonging to this task, the set of input topic partitions and changelog topic partitions
+ * for all logged state stores, and the rack ids of all replicas of each topic partition
+ * in the task.
+ */
+public interface TaskInfo {
+ TaskId id();
+
+ boolean isStateful();
+
+ Set<String> stateStoreNames();
+
+ Set<TopicPartition> sourceTopicPartitions();
+
+ Set<TopicPartition> changelogTopicPartitions();
Review Comment:
We should add javadocs to these getters. Will be trivial for most of them but it's important to note that this set contains partitions for source changelog topics as well, and that source changelog topic partitions can be identified by checking if they exist in both this and the source topic partitions set
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
+ freshTopicPartitionInfo.forEach((topic, partitionInfos) -> {
+ partitionInfos.forEach(partitionInfo -> {
+ final int partition = partitionInfo.partition();
+ final TopicPartition topicPartition = new TopicPartition(topic, partition);
+ final List<Node> replicas = partitionInfo.replicas();
+ if (replicas == null || replicas.isEmpty()) {
+ LOG.error("No replicas found for topic partition {}: {}", topic, partition);
+ return;
+ }
+
+ final Set<String> racks = replicas.stream().filter(Node::hasRack).map(Node::rack).collect(
+ Collectors.toSet());
+ racksForTopicPartition.computeIfAbsent(topicPartition, k -> new HashSet<>());
+ racksForTopicPartition.get(topicPartition).addAll(racks);
+ });
+ });
+
+ return racksForTopicPartition;
+ }
+
+ public static Set<String> topicsWithStaleMetadata(final Cluster cluster, final Set<TopicPartition> topicPartitions) {
+ final Set<String> topicsWithStaleMetadata = new HashSet<>();
+ for (final TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo = cluster.partition(topicPartition);
+ if (partitionInfo == null) {
+ LOG.error("TopicPartition {} doesn't exist in cluster", topicPartition);
+ continue;
Review Comment:
It's not an error for a topic to not be included in the `Cluster`, even source topics might not exist here if they had to be created by the assignor during the rebalance, since the `Cluster` metadata represents the state of the cluster when this rebalance/assignment first began.
Since the point of this method seems to be to collect topics with missing metadata that we'll need to look up via a DescribeTopics request, the ones for which `cluster.partition(topicPartition)` returns `null` are exactly the ones that should be returned by this method.
In fact I'd go ahead and remove everything past this line as well, this method should focus only on differentiating topics with missing metadata from ones we already have the info for. If the `Cluster` has metadata for this partition but the `replicas` set is missing/empty, then there's something wrong with this partition, and calling DescribeTopics probably won't help
Let's rename this to `#topicsWithMissingMetadata` while we're at it
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
+ freshTopicPartitionInfo.forEach((topic, partitionInfos) -> {
+ partitionInfos.forEach(partitionInfo -> {
+ final int partition = partitionInfo.partition();
+ final TopicPartition topicPartition = new TopicPartition(topic, partition);
+ final List<Node> replicas = partitionInfo.replicas();
+ if (replicas == null || replicas.isEmpty()) {
+ LOG.error("No replicas found for topic partition {}: {}", topic, partition);
+ return;
Review Comment:
nit: can you factor the lambda out into a separate method? I was really confused by this empty return for a while until I realized it wasn't returning from the `getRacksForTopicPartition` method, just the lambda inside this loop
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -1974,6 +1974,15 @@ public Set<InternalTopicConfig> nonSourceChangelogTopics() {
return topicConfigs;
}
+ /**
+ *
+ * @return the set of changelog topics, which includes both source changelog topics and non
+ * source changelog topics.
+ */
+ public Set<InternalTopicConfig> stateChangelogTopics() {
Review Comment:
nit: I'd just call this changelogTopics which I think helps make it obvious that this is the super-set of the #sourceChangelogTopics and #nonSourceChangelogTopics APIs
(you can rename the field itself as well but don't have to, that's up to you)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
Review Comment:
It's not a huge deal but if we have time left at the end it might make sense to condense this into a single call where we describe all the topics in one go rather than making a separate request for the source topics and changelogs
But it really isn't a big deal because in general, after the first rebalance, all the source topics should have been created and we really will do only one call since only the changelogs will be unknown
on that note, can you check to make sure this skips the actual DescribeTopics request if the set of topics to describe is empty? Like does it end up making a call with the admin client? If not then we should guard this with a `if (!topicsToDescribe.isEmpty)` (or we can just add this check anyways to be safe)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
Review Comment:
nit: make this final and add private constructor so it's clear this is just a static utils class
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
Review Comment:
if all we need is the topic names from the #stateChangelogTopics API then let's just have it return that directly. You should be able to just return the #keySet of that `stateChangelogTopics` map to get a Set<String> with the topic names right away
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1607404895
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -1974,6 +1974,15 @@ public Set<InternalTopicConfig> nonSourceChangelogTopics() {
return topicConfigs;
}
+ /**
+ *
+ * @return the set of changelog topics, which includes both source changelog topics and non
+ * source changelog topics.
+ */
+ public Set<String> changelogTopics() {
+ return Collections.unmodifiableSet(new HashSet<>(stateChangelogTopics.keySet()));
Review Comment:
I think you can skip the new HashSet step, that's pretty much redundant with the unmodifiableSet and since we don't plan on modifying the returned set, it's better to just wrap the keySet directly to save a bunch of unnecessary copying
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman merged PR #15972:
URL: https://github.com/apache/kafka/pull/15972
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1604122125
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskInfo.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.assignment;
+
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.assignment.TaskInfo;
+
+public class DefaultTaskInfo implements TaskInfo {
+
+ private final TaskId id;
+ private final boolean isStateful;
+ private final Map<TopicPartition, Set<String>> partitionToRackIds;
+ private final Set<String> stateStoreNames;
+ private final Set<TopicPartition> inputTopicPartitions;
+ private final Set<TopicPartition> changelogTopicPartitions;
+
+ public DefaultTaskInfo(final TaskId id,
+ final boolean isStateful,
+ final Map<TopicPartition, Set<String>> partitionToRackIds,
+ final Set<String> stateStoreNames,
+ final Set<TopicPartition> inputTopicPartitions,
+ final Set<TopicPartition> changelogTopicPartitions) {
+ this.id = id;
+ this.partitionToRackIds = unmodifiableMap(partitionToRackIds);
+ this.isStateful = isStateful;
+ this.stateStoreNames = unmodifiableSet(stateStoreNames);
+ this.inputTopicPartitions = unmodifiableSet(inputTopicPartitions);
+ this.changelogTopicPartitions = unmodifiableSet(changelogTopicPartitions);
+ }
+
+ public static DefaultTaskInfo of(final TaskId taskId,
Review Comment:
since this is an internal API, you can just have a normal public constructor. The static constructor thing is only for public classes where we want to make a "nice looking" fluent API
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskInfo.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.assignment;
+
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.assignment.TaskInfo;
+
+public class DefaultTaskInfo implements TaskInfo {
+
+ private final TaskId id;
+ private final boolean isStateful;
+ private final Map<TopicPartition, Set<String>> partitionToRackIds;
+ private final Set<String> stateStoreNames;
+ private final Set<TopicPartition> inputTopicPartitions;
+ private final Set<TopicPartition> changelogTopicPartitions;
+
+ public DefaultTaskInfo(final TaskId id,
+ final boolean isStateful,
+ final Map<TopicPartition, Set<String>> partitionToRackIds,
+ final Set<String> stateStoreNames,
+ final Set<TopicPartition> inputTopicPartitions,
+ final Set<TopicPartition> changelogTopicPartitions) {
+ this.id = id;
+ this.partitionToRackIds = unmodifiableMap(partitionToRackIds);
+ this.isStateful = isStateful;
+ this.stateStoreNames = unmodifiableSet(stateStoreNames);
+ this.inputTopicPartitions = unmodifiableSet(inputTopicPartitions);
+ this.changelogTopicPartitions = unmodifiableSet(changelogTopicPartitions);
+ }
+
+ public static DefaultTaskInfo of(final TaskId taskId,
+ final boolean isStateful,
+ final ValueMapper<TopicPartition, String> previousOwnerForPartition,
+ final Map<String, Optional<String>> rackForConsumer,
+ final Map<TaskId, Set<TopicPartition>> inputPartitionsForTask,
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask) {
+
+ final Set<TopicPartition> inputPartitions = inputPartitionsForTask.get(taskId);
+ final Set<TopicPartition> changelogPartitions = changelogPartitionsForTask.get(taskId);
+ final Map<TopicPartition, Set<String>> racksForPartition = new HashMap<>();
+
+ inputPartitions.forEach(partition -> {
+ racksForPartition.computeIfAbsent(partition, k -> new HashSet<>());
+ final String consumer = previousOwnerForPartition.apply(partition);
+ final Optional<String> rack = rackForConsumer.get(consumer);
+ rack.ifPresent(s -> racksForPartition.get(partition).add(s));
+ });
+
+ changelogPartitions.forEach(partition -> {
+ racksForPartition.computeIfAbsent(partition, k -> new HashSet<>());
+ final String consumer = previousOwnerForPartition.apply(partition);
+ final Optional<String> rack = rackForConsumer.get(consumer);
+ rack.ifPresent(s -> racksForPartition.get(partition).add(s));
+ });
+
+ final Set<String> stateStoreNames = new HashSet<>();
+ return new DefaultTaskInfo(
+ taskId,
+ isStateful, // All standby tasks are stateful.
Review Comment:
Ah, another thing to note here is that this class corresponds to a "logical task", not a "physical one". I just made up those terms but hopefully this will make sense: a "physical" task can be active or standby and represents an actual task that was assigned to a client and will be running on that client, where a "logical" task is just the metadata corresponding to that task id. Where a "task id" logically represents a combination of subtopology (grouping of processors) and partition number. So a logical task doesn't have a concept of active vs standby because it's just metadata, this class is basically telling the assignor which tasks exist in this application. The assignor then has to create a set of physical tasks to actually be assigned, basically one active task and however many standby tasks for each "logical task"
Hope that didn't make things more confusing...anyways this comment isn't incorrect, but it doesn't exactly apply in this context. The "isStateful" is just metadata related to whether it has state stores in this subtopology (I'll tell you how to get this info later, or you can even compute it based on `stateStoresNames#isEmpty`)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskInfo.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.assignment;
+
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.assignment.TaskInfo;
+
+public class DefaultTaskInfo implements TaskInfo {
+
+ private final TaskId id;
+ private final boolean isStateful;
+ private final Map<TopicPartition, Set<String>> partitionToRackIds;
+ private final Set<String> stateStoreNames;
+ private final Set<TopicPartition> inputTopicPartitions;
+ private final Set<TopicPartition> changelogTopicPartitions;
+
+ public DefaultTaskInfo(final TaskId id,
+ final boolean isStateful,
+ final Map<TopicPartition, Set<String>> partitionToRackIds,
+ final Set<String> stateStoreNames,
+ final Set<TopicPartition> inputTopicPartitions,
+ final Set<TopicPartition> changelogTopicPartitions) {
+ this.id = id;
+ this.partitionToRackIds = unmodifiableMap(partitionToRackIds);
+ this.isStateful = isStateful;
+ this.stateStoreNames = unmodifiableSet(stateStoreNames);
+ this.inputTopicPartitions = unmodifiableSet(inputTopicPartitions);
+ this.changelogTopicPartitions = unmodifiableSet(changelogTopicPartitions);
+ }
+
+ public static DefaultTaskInfo of(final TaskId taskId,
+ final boolean isStateful,
+ final ValueMapper<TopicPartition, String> previousOwnerForPartition,
+ final Map<String, Optional<String>> rackForConsumer,
+ final Map<TaskId, Set<TopicPartition>> inputPartitionsForTask,
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask) {
+
+ final Set<TopicPartition> inputPartitions = inputPartitionsForTask.get(taskId);
+ final Set<TopicPartition> changelogPartitions = changelogPartitionsForTask.get(taskId);
+ final Map<TopicPartition, Set<String>> racksForPartition = new HashMap<>();
+
+ inputPartitions.forEach(partition -> {
+ racksForPartition.computeIfAbsent(partition, k -> new HashSet<>());
+ final String consumer = previousOwnerForPartition.apply(partition);
+ final Optional<String> rack = rackForConsumer.get(consumer);
+ rack.ifPresent(s -> racksForPartition.get(partition).add(s));
+ });
Review Comment:
Ah, this is not computing the right rack id -- this would be the rack id of the KafkaStreams node that had this partition assigned during the last rebalance. What we want is the rack.id of the broker node(s) that host this partition. This is going to be a bit complex so let's chat online (ditto for the changelogPartitions rack info as well)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskInfo.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.assignment;
+
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.assignment.TaskInfo;
+
+public class DefaultTaskInfo implements TaskInfo {
+
+ private final TaskId id;
+ private final boolean isStateful;
+ private final Map<TopicPartition, Set<String>> partitionToRackIds;
+ private final Set<String> stateStoreNames;
+ private final Set<TopicPartition> inputTopicPartitions;
+ private final Set<TopicPartition> changelogTopicPartitions;
+
+ public DefaultTaskInfo(final TaskId id,
+ final boolean isStateful,
+ final Map<TopicPartition, Set<String>> partitionToRackIds,
+ final Set<String> stateStoreNames,
+ final Set<TopicPartition> inputTopicPartitions,
+ final Set<TopicPartition> changelogTopicPartitions) {
+ this.id = id;
+ this.partitionToRackIds = unmodifiableMap(partitionToRackIds);
+ this.isStateful = isStateful;
+ this.stateStoreNames = unmodifiableSet(stateStoreNames);
+ this.inputTopicPartitions = unmodifiableSet(inputTopicPartitions);
+ this.changelogTopicPartitions = unmodifiableSet(changelogTopicPartitions);
+ }
+
+ public static DefaultTaskInfo of(final TaskId taskId,
+ final boolean isStateful,
+ final ValueMapper<TopicPartition, String> previousOwnerForPartition,
+ final Map<String, Optional<String>> rackForConsumer,
+ final Map<TaskId, Set<TopicPartition>> inputPartitionsForTask,
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask) {
+
+ final Set<TopicPartition> inputPartitions = inputPartitionsForTask.get(taskId);
Review Comment:
if this is the only place where we use the `inputPartitionsForTask`/`changelogPartitionsForTask` map, let's just pass in the `inputPartitions` & `changelogPartitions` sets directly
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "apourchet (via GitHub)" <gi...@apache.org>.
apourchet commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1606917357
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
+ freshTopicPartitionInfo.forEach((topic, partitionInfos) -> {
+ partitionInfos.forEach(partitionInfo -> {
+ final int partition = partitionInfo.partition();
+ final TopicPartition topicPartition = new TopicPartition(topic, partition);
+ final List<Node> replicas = partitionInfo.replicas();
+ if (replicas == null || replicas.isEmpty()) {
+ LOG.error("No replicas found for topic partition {}: {}", topic, partition);
+ return;
+ }
+
+ final Set<String> racks = replicas.stream().filter(Node::hasRack).map(Node::rack).collect(
+ Collectors.toSet());
+ racksForTopicPartition.computeIfAbsent(topicPartition, k -> new HashSet<>());
+ racksForTopicPartition.get(topicPartition).addAll(racks);
+ });
+ });
+
+ return racksForTopicPartition;
+ }
+
+ public static Set<String> topicsWithStaleMetadata(final Cluster cluster, final Set<TopicPartition> topicPartitions) {
+ final Set<String> topicsWithStaleMetadata = new HashSet<>();
+ for (final TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo = cluster.partition(topicPartition);
+ if (partitionInfo == null) {
+ LOG.error("TopicPartition {} doesn't exist in cluster", topicPartition);
+ continue;
Review Comment:
This isn't how the current rack aware code works:
``` for (final TopicPartition topicPartition : topicPartitions) {
final PartitionInfo partitionInfo = fullMetadata.partition(topicPartition);
if (partitionInfo == null) {
log.error("TopicPartition {} doesn't exist in cluster", topicPartition);
return false;
}
final Node[] replica = partitionInfo.replicas();
if (replica == null || replica.length == 0) {
topicsToDescribe.add(topicPartition.topic());
continue;
}
for (final Node node : replica) {
if (node.hasRack()) {
racksForPartition.computeIfAbsent(topicPartition, k -> new HashSet<>()).add(node.rack());
} else {
log.warn("Node {} for topic partition {} doesn't have rack", node, topicPartition);
return false;
}
}
}
```
Above is the logic of `populateTopicsToDescribe`, which evidently uses the `(replica == null || replica.length == 0)` condition to decide to fetch further topic information, yet considers `cluster.partition(topicPartition)` to be an error, which causes the RackAwareAssignor to be turned off entirely.
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1605656530
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
+ sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
+ changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
}
+ final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
+ partitionGrouper.partitionGroups(sourceTopicsByGroup, cluster);
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask =
+ partitionGrouper.partitionGroups(changelogTopicsByGroup, cluster);
+
+ final Set<TaskId> logicalTaskIds = new HashSet<>();
+ final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
+ sourcePartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ sourceTopicPartitions.addAll(partitions);
+ });
+ final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
+ changelogPartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ changelogTopicPartitions.addAll(partitions);
+ });
+
+ final Map<TopicPartition, Set<String>> racksForSourcePartitions = RackUtils.getRacksForTopicPartition(
+ cluster, internalTopicManager, sourceTopicPartitions, false);
+ final Map<TopicPartition, Set<String>> racksForChangelogPartitions = RackUtils.getRacksForTopicPartition(
Review Comment:
Since the rack info is nontrivial to compute and always makes a remote call (which can take a long time and even time out or otherwise fail) and not every assignor/app will actually use it I'm thinking maybe we should try to initialize it lazily, only if/when the user actually requests the rack info
I'm totally happy to push that into a followup PR to keep the scope well-defined for now, so don't worry about it for now. We'd still need everything you implemented here and would just be moving it around and/or subbing in function pointers instead of passing around data strucutres directly, so it shouldn't have any impact on how this PR goes. Just wanted to make a note so I don't forget
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1605659120
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
+ sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
+ changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
}
+ final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
+ partitionGrouper.partitionGroups(sourceTopicsByGroup, cluster);
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask =
+ partitionGrouper.partitionGroups(changelogTopicsByGroup, cluster);
+
+ final Set<TaskId> logicalTaskIds = new HashSet<>();
+ final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
+ sourcePartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ sourceTopicPartitions.addAll(partitions);
+ });
+ final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
+ changelogPartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
Review Comment:
Note that we'll also want to deduplicate the source-changelog partitions for the rack id computation. We should include them in the source topics/remove them from the changelog topics passed into the `#getRacksForTopicPartitions` call. Of course we still need the changelogTopicPartitions as well, so we'll want a third set of `nonSourceChangelogTopicPartitions` that's specifically for the rack id computation.
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1605660608
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -472,23 +482,74 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
*
* @param clientMetadataMap the map of process id to client metadata used to build an immutable
* {@code ApplicationState}
- * @param statefulTasks the set of {@code TaskId} that correspond to all the stateful
- * tasks that need to be reassigned.
* @return The {@code ApplicationState} needed by the TaskAssigner to compute new task
* assignments.
*/
- private ApplicationState buildApplicationState(final Map<UUID, ClientMetadata> clientMetadataMap,
- final Set<TaskId> statefulTasks) {
- final Set<TaskId> statelessTasks = new HashSet<>();
- for (final Map.Entry<UUID, ClientMetadata> clientEntry : clientMetadataMap.entrySet()) {
- final ClientState clientState = clientEntry.getValue().state;
- statelessTasks.addAll(clientState.statelessActiveTasks());
+ private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
+ final Map<UUID, ClientMetadata> clientMetadataMap,
+ final Map<Subtopology, TopicsInfo> topicGroups,
+ final Cluster cluster) {
+ final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
+ final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
+ for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
+ final Set<String> sourceTopics = entry.getValue().sourceTopics;
+ final Set<String> changelogTopics = entry.getValue().stateChangelogTopics()
+ .stream().map(t -> t.name).collect(Collectors.toSet());
+ sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
+ changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
}
+ final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
+ partitionGrouper.partitionGroups(sourceTopicsByGroup, cluster);
+ final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask =
+ partitionGrouper.partitionGroups(changelogTopicsByGroup, cluster);
+
+ final Set<TaskId> logicalTaskIds = new HashSet<>();
+ final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
+ sourcePartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
+ sourceTopicPartitions.addAll(partitions);
+ });
+ final Set<TopicPartition> changelogTopicPartitions = new HashSet<>();
+ changelogPartitionsForTask.forEach((taskId, partitions) -> {
+ logicalTaskIds.add(taskId);
Review Comment:
Sorry for the wall of text 😅 It might not seem like a huge deal but if it's an app with only source-changelog partitions, then doing this will save the assignor from having to make any DescribeTopics request since there are no non-source changelogs.
And yes, apps with only source changelogs do exist, they're pretty common for certain kinds of table-based processing (and especially apps that make heavy use of IQ). And saving a remote fetch is actually a pretty big deal, doing them in the middle of an assignment makes the rebalance vulnerable to timing out, especially when brokers are under heavy load or the app is experiencing rebalancing issues to begin with
--
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-15045: (KIP-924 pt. 5) Add rack information to ApplicationState [kafka]
Posted by "apourchet (via GitHub)" <gi...@apache.org>.
apourchet commented on code in PR #15972:
URL: https://github.com/apache/kafka/pull/15972#discussion_r1606917357
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.assignment;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+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.TopicPartitionInfo;
+import org.apache.kafka.streams.processor.internals.InternalTopicManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(RackUtils.class);
+
+ public static Map<TopicPartition, Set<String>> getRacksForTopicPartition(final Cluster cluster,
+ final InternalTopicManager internalTopicManager,
+ final Set<TopicPartition> topicPartitions,
+ final boolean isChangelog) {
+ final Set<String> topicsToDescribe = new HashSet<>();
+ if (isChangelog) {
+ topicsToDescribe.addAll(topicPartitions.stream().map(TopicPartition::topic).collect(
+ Collectors.toSet()));
+ } else {
+ topicsToDescribe.addAll(topicsWithStaleMetadata(cluster, topicPartitions));
+ }
+
+ final Set<TopicPartition> topicsWithUpToDateMetadata = topicPartitions.stream()
+ .filter(partition -> !topicsToDescribe.contains(partition.topic()))
+ .collect(Collectors.toSet());
+ final Map<TopicPartition, Set<String>> racksForTopicPartition = knownRacksForPartition(
+ cluster, topicsWithUpToDateMetadata);
+
+ final Map<String, List<TopicPartitionInfo>> freshTopicPartitionInfo =
+ describeTopics(internalTopicManager, topicsToDescribe);
+ freshTopicPartitionInfo.forEach((topic, partitionInfos) -> {
+ partitionInfos.forEach(partitionInfo -> {
+ final int partition = partitionInfo.partition();
+ final TopicPartition topicPartition = new TopicPartition(topic, partition);
+ final List<Node> replicas = partitionInfo.replicas();
+ if (replicas == null || replicas.isEmpty()) {
+ LOG.error("No replicas found for topic partition {}: {}", topic, partition);
+ return;
+ }
+
+ final Set<String> racks = replicas.stream().filter(Node::hasRack).map(Node::rack).collect(
+ Collectors.toSet());
+ racksForTopicPartition.computeIfAbsent(topicPartition, k -> new HashSet<>());
+ racksForTopicPartition.get(topicPartition).addAll(racks);
+ });
+ });
+
+ return racksForTopicPartition;
+ }
+
+ public static Set<String> topicsWithStaleMetadata(final Cluster cluster, final Set<TopicPartition> topicPartitions) {
+ final Set<String> topicsWithStaleMetadata = new HashSet<>();
+ for (final TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo = cluster.partition(topicPartition);
+ if (partitionInfo == null) {
+ LOG.error("TopicPartition {} doesn't exist in cluster", topicPartition);
+ continue;
Review Comment:
This isn't how the current rack aware code works:
``` for (final TopicPartition topicPartition : topicPartitions) {
final PartitionInfo partitionInfo = fullMetadata.partition(topicPartition);
if (partitionInfo == null) {
log.error("TopicPartition {} doesn't exist in cluster", topicPartition);
return false;
}
final Node[] replica = partitionInfo.replicas();
if (replica == null || replica.length == 0) {
topicsToDescribe.add(topicPartition.topic());
continue;
}
for (final Node node : replica) {
if (node.hasRack()) {
racksForPartition.computeIfAbsent(topicPartition, k -> new HashSet<>()).add(node.rack());
} else {
log.warn("Node {} for topic partition {} doesn't have rack", node, topicPartition);
return false;
}
}
}
```
Above is the logic of `populateTopicsToDescribe`, which evidently uses the `(replica == null || replica.length == 0)` condition to decide to fetch further topic information, yet considers `cluster.partition(topicPartition)` to be an error.
--
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