ableegoldman commented on code in PR #16034:
URL: https://github.com/apache/kafka/pull/16034#discussion_r1610769106
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -38,26 +40,37 @@ public final class RackUtils {
private RackUtils() { }
- 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(topicsWithMissingMetadata(cluster,
topicPartitions));
- }
+ public static void annotateWithTopicPartitionsWithRackInfo(final Cluster
cluster,
+ final
InternalTopicManager internalTopicManager,
Review Comment:
nit: fix the formatting/indentation (although this will change when you
change the method name anyway)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackUtils.java:
##########
@@ -38,26 +40,37 @@ public final class RackUtils {
private RackUtils() { }
- 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(topicsWithMissingMetadata(cluster,
topicPartitions));
- }
+ public static void annotateWithTopicPartitionsWithRackInfo(final Cluster
cluster,
Review Comment:
this doesn't sound quite right, should it be
`annotateTopicPartitionsWithRackInfo`? (ie extra "with" )
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java:
##########
@@ -513,50 +515,45 @@ private ApplicationState buildApplicationState(final
TopologyMetadata topologyMe
+ "tasks for source topics vs
changelog topics.");
}
- final Set<TopicPartition> sourceTopicPartitions = new HashSet<>();
- final Set<TopicPartition> nonSourceChangelogTopicPartitions = new
HashSet<>();
- for (final Map.Entry<TaskId, Set<TopicPartition>> entry :
sourcePartitionsForTask.entrySet()) {
- final TaskId taskId = entry.getKey();
- final Set<TopicPartition> taskSourcePartitions = entry.getValue();
- final Set<TopicPartition> taskChangelogPartitions =
changelogPartitionsForTask.get(taskId);
- final Set<TopicPartition> taskNonSourceChangelogPartitions = new
HashSet<>(taskChangelogPartitions);
- taskNonSourceChangelogPartitions.removeAll(taskSourcePartitions);
-
- sourceTopicPartitions.addAll(taskSourcePartitions);
-
nonSourceChangelogTopicPartitions.addAll(taskNonSourceChangelogPartitions);
- }
+ final Set<TaskId> logicalTaskIds =
unmodifiableSet(sourcePartitionsForTask.keySet());
+ final Set<DefaultTaskTopicPartition> allTopicPartitions = new
HashSet<>();
+ final Map<TaskId, Set<TaskTopicPartition>> topicPartitionsForTask =
new HashMap<>();
+ logicalTaskIds.forEach(taskId -> {
+ final Set<TaskTopicPartition> topicPartitions = new HashSet<>();
+
+ for (final TopicPartition topicPartition :
sourcePartitionsForTask.get(taskId)) {
+ final boolean isSource = true;
+ final boolean isChangelog =
changelogPartitionsForTask.get(taskId).contains(topicPartition);
+ final DefaultTaskTopicPartition racklessTopicPartition = new
DefaultTaskTopicPartition(
+ topicPartition, isSource, isChangelog, null);
+ allTopicPartitions.add(racklessTopicPartition);
+ topicPartitionsForTask.get(taskId).add(racklessTopicPartition);
Review Comment:
both here and for the changelog loop below
```suggestion
topicPartitions.add(racklessTopicPartition);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
