apourchet commented on code in PR #16034:
URL: https://github.com/apache/kafka/pull/16034#discussion_r1610816330
##########
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:
Good catch!
--
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
