frankvicky commented on code in PR #18989:
URL: https://github.com/apache/kafka/pull/18989#discussion_r1969053382
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2592,6 +2606,32 @@ class KafkaApis(val requestChannel: RequestChannel,
response.groups.addAll(results)
}
+ // Clients are not allowed to see topics that are not authorized for
Describe.
+ val topicsToCheck = new mutable.HashSet[String]
+ response.groups.forEach(_.members.forEach { member =>
+ List(member.assignment, member.targetAssignment).foreach {
assignment =>
+ assignment.topicPartitions.asScala.foreach { tp =>
+ topicsToCheck += tp.topicName
+ }
+ }
+ })
Review Comment:
This chunck of code is not easy to read. Could you refactor this one?
For example:
```scala
// Clients are not allowed to see topics that are not authorized
for Describe.
val topicsToCheck = response.groups.stream()
.flatMap(group => group.members.stream)
.flatMap(member => util.stream.Stream.of(member.assignment,
member.targetAssignment))
.flatMap(assignment => assignment.topicPartitions.stream)
.map(topicPartition => topicPartition.topicName)
.collect(Collectors.toSet[String])
.asScala
```
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2592,6 +2606,32 @@ class KafkaApis(val requestChannel: RequestChannel,
response.groups.addAll(results)
}
+ // Clients are not allowed to see topics that are not authorized for
Describe.
+ val topicsToCheck = new mutable.HashSet[String]
+ response.groups.forEach(_.members.forEach { member =>
+ List(member.assignment, member.targetAssignment).foreach {
assignment =>
+ assignment.topicPartitions.asScala.foreach { tp =>
+ topicsToCheck += tp.topicName
+ }
+ }
+ })
+ val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
+ topicsToCheck)(identity)
+ val updatedGroups = response.groups.asScala.map { group =>
+ if (group.members.asScala.exists(member =>
+ List(member.assignment,
member.targetAssignment).exists(assignment =>
+ assignment.topicPartitions.asScala.exists(tp =>
!authorizedTopics.contains(tp.topicName))))) {
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(group.groupId)
+ .setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
+ .setErrorMessage("The group has described topic(s) that the
client is not authorized to describe.")
+ .setMembers(List.empty.asJava)
+ } else {
+ group
+ }
+ }.asJava
Review Comment:
ditto.
For example:
```scala
val updatedGroups = response.groups.stream().map { group =>
val hasUnauthorizedTopic = group.members.stream()
.flatMap(member => util.stream.Stream.of(member.assignment,
member.targetAssignment))
.flatMap(assignment => assignment.topicPartitions.stream())
.map(tp => tp.topicName)
.anyMatch(topicName => !authorizedTopics.contains(topicName))
if (hasUnauthorizedTopic) {
new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupId(group.groupId)
.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
.setErrorMessage("The group has described topic(s) that the
client is not authorized to describe.")
.setMembers(List.empty.asJava)
} else {
group
}
}.collect(Collectors.toList[ConsumerGroupDescribeResponseData.DescribedGroup])
```
--
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
