dajac commented on code in PR #15205: URL: https://github.com/apache/kafka/pull/15205#discussion_r1463681361
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java: ########## @@ -578,26 +578,38 @@ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember( private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap( Map<Uuid, Set<Integer>> partitions, - Map<String, TopicMetadata> subscriptionMetadata + TopicsImage topicsImage ) { - return partitions.entrySet().stream().map( - item -> new ConsumerGroupDescribeResponseData.TopicPartitions() - .setTopicId(item.getKey()) - .setTopicName(lookupTopicNameById(item.getKey(), subscriptionMetadata)) - .setPartitions(new ArrayList<>(item.getValue())) - ).collect(Collectors.toList()); + List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitions = new ArrayList<>(); + for (Map.Entry<Uuid, Set<Integer>> entry : partitions.entrySet()) { + Uuid topicId = entry.getKey(); + Set<Integer> partitionSet = partitions.get(topicId); +// partitions.forEach((topicId, partitionSet) -> { + String topicName = lookupTopicNameById(topicId, topicsImage); + if (topicName != null) { + topicPartitions.add(new ConsumerGroupDescribeResponseData.TopicPartitions() + .setTopicId(topicId) + .setTopicName(topicName) + .setPartitions(new ArrayList<>(partitionSet))); + } else { + // When the topic has been deleted and the group/member hasn't updated, + // directly remove the topic from the assignment. + partitions.remove(topicId, partitionSet); Review Comment: We should not do this. We should only not add it to the reponse. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java: ########## @@ -578,26 +578,38 @@ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember( private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap( Map<Uuid, Set<Integer>> partitions, - Map<String, TopicMetadata> subscriptionMetadata + TopicsImage topicsImage ) { - return partitions.entrySet().stream().map( - item -> new ConsumerGroupDescribeResponseData.TopicPartitions() - .setTopicId(item.getKey()) - .setTopicName(lookupTopicNameById(item.getKey(), subscriptionMetadata)) - .setPartitions(new ArrayList<>(item.getValue())) - ).collect(Collectors.toList()); + List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitions = new ArrayList<>(); + for (Map.Entry<Uuid, Set<Integer>> entry : partitions.entrySet()) { Review Comment: nit: How about using `partitions.foreach`? ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java: ########## @@ -17,10 +17,11 @@ package org.apache.kafka.coordinator.group.consumer; Review Comment: Should we add a unit test here to verify that non-existent partitions are not provided in the response? ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java: ########## @@ -578,26 +578,38 @@ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember( private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap( Map<Uuid, Set<Integer>> partitions, - Map<String, TopicMetadata> subscriptionMetadata + TopicsImage topicsImage ) { - return partitions.entrySet().stream().map( - item -> new ConsumerGroupDescribeResponseData.TopicPartitions() - .setTopicId(item.getKey()) - .setTopicName(lookupTopicNameById(item.getKey(), subscriptionMetadata)) - .setPartitions(new ArrayList<>(item.getValue())) - ).collect(Collectors.toList()); + List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitions = new ArrayList<>(); + for (Map.Entry<Uuid, Set<Integer>> entry : partitions.entrySet()) { + Uuid topicId = entry.getKey(); + Set<Integer> partitionSet = partitions.get(topicId); +// partitions.forEach((topicId, partitionSet) -> { Review Comment: nit: Could we remove it? -- 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