You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dongnuo123 (via GitHub)" <gi...@apache.org> on 2024/01/16 23:31:21 UTC
[PR] MINOR: populate TopicName in ConsumerGroupDescribe [kafka]
dongnuo123 opened a new pull request, #15205:
URL: https://github.com/apache/kafka/pull/15205
The patch populates the topic name of `ConsumerGroupDescribeResponseData.TopicPartitions` with the corresponding topic id in `ConsumerGroupDescribe`.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: populate TopicName in ConsumerGroupDescribe [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #15205:
URL: https://github.com/apache/kafka/pull/15205
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: populate TopicName in ConsumerGroupDescribe [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15205:
URL: https://github.com/apache/kafka/pull/15205#discussion_r1458581509
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java:
##########
@@ -571,15 +577,29 @@ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember(As
}
private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap(
- Map<Uuid, Set<Integer>> partitions
+ Map<Uuid, Set<Integer>> partitions,
+ Map<String, TopicMetadata> subscriptionMetadata
) {
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());
}
+ private static String lookupTopicNameById(
+ Uuid topicId,
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ for (TopicMetadata topicMetadata : subscriptionMetadata.values()) {
+ if (topicId.equals(topicMetadata.id())) {
+ return topicMetadata.name();
+ }
+ }
+ throw UNKNOWN_TOPIC_ID.exception("The subscription metadata does not contain " + topicId);
Review Comment:
I am not sure about this. It means that when a topic is delete and the group has not been updated yet, it won't be possible to describe it. An alternative would be to directly remove the topic from the assignment. It will be eventually removed anyway. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java:
##########
@@ -571,15 +577,29 @@ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember(As
}
private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap(
- Map<Uuid, Set<Integer>> partitions
+ Map<Uuid, Set<Integer>> partitions,
+ Map<String, TopicMetadata> subscriptionMetadata
) {
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());
}
+ private static String lookupTopicNameById(
+ Uuid topicId,
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ for (TopicMetadata topicMetadata : subscriptionMetadata.values()) {
+ if (topicId.equals(topicMetadata.id())) {
+ return topicMetadata.name();
+ }
+ }
Review Comment:
This is pretty bad, isn't it? I was wondering if we could use the MetadataImage instead. It would allow us to get the name based on the id.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: populate TopicName in ConsumerGroupDescribe [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15205:
URL: https://github.com/apache/kafka/pull/15205#discussion_r1463772765
##########
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:
Yeah, we have that in `testAsConsumerGroupDescribeWithTopicNameNotFound`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: populate TopicName in ConsumerGroupDescribe [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org