mjsax commented on code in PR #23006:
URL: https://github.com/apache/kafka/pull/23006#discussion_r3700737746
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java:
##########
@@ -74,10 +74,14 @@ public static
StreamsGroupHeartbeatResponseData.EndpointToPartitions endpointToP
entry -> entry.getValue().keySet()
));
Map<String, Set<Integer>> standbyTasks =
streamsGroupMember.assignedTasks().standbyTasks();
+ Map<String, Set<Integer>> warmupTasks =
streamsGroupMember.assignedTasks().warmupTasks();
endpointToPartitions.setUserEndpoint(responseEndpoint);
Map<String, ConfiguredSubtopology> configuredSubtopologies =
streamsGroup.configuredTopology().flatMap(ConfiguredTopology::subtopologies).get();
List<StreamsGroupHeartbeatResponseData.TopicPartition>
activeTopicPartitions = topicPartitions(activeTasks, configuredSubtopologies,
metadataImage);
List<StreamsGroupHeartbeatResponseData.TopicPartition>
standbyTopicPartitions = topicPartitions(standbyTasks, configuredSubtopologies,
metadataImage);
+ List<StreamsGroupHeartbeatResponseData.TopicPartition>
warmupTopicPartitions = topicPartitions(warmupTasks, configuredSubtopologies,
metadataImage);
+ // Warm-up tasks are executed as standby tasks on the client, so they
are reported as standby partitions, as in the classic protocol.
+ standbyTopicPartitions.addAll(warmupTopicPartitions);
Review Comment:
I don't think this is correct. Instead of merging the result of
`topicPartitions(...)`, we need to merge `standbyTasks` and `warmupTasks` and
pass the merged set into `topicPartitions(...)`.
The problem we are hitting is about "uneven" tasks, ie, if we have a task T
reading from topic A (with 5 partitions) and topic B (with 2 partition), only
task 0_0, and 0_1 read from both topics (ie, they read topic-partitions A-0 and
B-0, and A-1 and B-1, respectively). However, for tasks 0_2, 0_3, 0_4, there is
no partitions for topic B and these task only read from topic A (ie,
topic-partition A-2, A-3, and A-4).
To account for cases like this, we "merge" order is important, and only
merging first is correct; otherwise, we compute the incorrect metadata.
(To be fair, Claude pointed it out -- Claud also found a pre-existing bug:
https://issues.apache.org/jira/browse/KAFKA-20885; I opened a PR for it already
https://github.com/apache/kafka/pull/23050, and we should merge my PR first,
and your PR can get rebased afterwards. W/o the other fix, this PR cannot write
correct unit tests for the warmup task case...)
--
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]
