mjsax commented on code in PR #23050:
URL: https://github.com/apache/kafka/pull/23050#discussion_r3707553256
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java:
##########
@@ -100,25 +99,32 @@ private static
List<StreamsGroupHeartbeatResponseData.TopicPartition> topicParti
return topicPartitionsForTasks;
}
- private static List<StreamsGroupHeartbeatResponseData.TopicPartition>
topicPartitionListForTask(final Set<Integer> taskSet,
-
final Set<String> topicNames,
-
final CoordinatorMetadataImage metadataImage) {
- return topicNames.stream().map(topic -> {
+ private static List<StreamsGroupHeartbeatResponseData.TopicPartition>
topicPartitionListForTask(
+ final Set<Integer> taskSet,
+ final Set<String> topicNames,
+ final CoordinatorMetadataImage metadataImage
+ ) {
+ List<StreamsGroupHeartbeatResponseData.TopicPartition>
topicPartitionsForTask = new ArrayList<>();
+ for (String topic : topicNames) {
Optional<CoordinatorMetadataImage.TopicMetadata> topicMetadata =
metadataImage.topicMetadata(topic);
if (topicMetadata.isEmpty()) {
throw new IllegalStateException("Topic " + topic + " not found
in metadata image");
}
int numPartitionsForTopic = topicMetadata.get().partitionCount();
- StreamsGroupHeartbeatResponseData.TopicPartition tp = new
StreamsGroupHeartbeatResponseData.TopicPartition();
- tp.setTopic(topic);
- List<Integer> tpPartitions = new ArrayList<>(taskSet);
- if (numPartitionsForTopic < taskSet.size()) {
- Collections.sort(tpPartitions);
- tp.setPartitions(tpPartitions.subList(0,
numPartitionsForTopic));
- } else {
- tp.setPartitions(tpPartitions);
+ // A subtopology has as many tasks as its source topic with the
most partitions. A source topic with
+ // fewer partitions therefore has no partition for the higher task
IDs, mirroring the client-side
+ // grouping in PartitionGrouper#partitionGroups.
+ List<Integer> partitions = taskSet.stream()
+ .filter(taskId -> taskId < numPartitionsForTopic)
Review Comment:
```suggestion
.filter(partitionId -> partitionId < numPartitionsForTopic)
```
--
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]
