KAFKA-18819 StreamsGroupHeartbeat API and StreamsGroupDescribe API check topic describe #19183
Conversation
…eck topic describe
github-actions
bot
added
triage
|
@lucasbru PTAL when you get a chance. |
| // - CLUSTER_AUTHORIZATION_FAILED (version 0+) | ||
| // - STREAMS_INVALID_TOPOLOGY (version 0+) | ||
| // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+) | ||
| // - STREAMS_TOPOLOGY_FENCED (version 0+) | ||
| // - TOPIC_AUTHORIZATION_FAILED (version 0+) |
| @@ -43,6 +43,7 @@ | |||
| * - {@link Errors#STREAMS_INVALID_TOPOLOGY} | |||
| * - {@link Errors#STREAMS_INVALID_TOPOLOGY_EPOCH} | |||
| * - {@link Errors#STREAMS_TOPOLOGY_FENCED} | |||
| * - {@link Errors#TOPIC_AUTHORIZATION_FAILED} | |||
| log.debug("`DescribeStreamsGroups` request for group id {} failed due to error {}", groupId.idValue, error); | ||
| // The topic auth response received on DescribeStreamsGroup is a generic one not including topic names, so we just pass it on unchanged here. |
There was a problem hiding this comment.
what do you mean by this comment? Also, this block also handles GROUP_AUTHORIZATION_FAILED, so it's weird that you're just talking about topic auth.
| .flatMap(subtopology => util.stream.Stream.of( | ||
| subtopology.sourceTopics, | ||
| subtopology.repartitionSinkTopics, | ||
| subtopology.repartitionSourceTopics.iterator.asScala.map(_.name).toList.asJava, |
There was a problem hiding this comment.
Couldn't you use .stream().map(_name) instead of converting between scala and java? It seems the extra collection is unncessary here.
Something like
Stream.concat(
subtopology.sourceTopics.stream,
subtopology.repartitionSinkTopics.stream,
subtopology.repartitionSourceTopics.stream.map(_.name),
....
)
| val updatedGroups = response.groups.stream.map { group => | ||
| val hasUnauthorizedTopic = group.topology.subtopologies.stream() | ||
| .flatMap(subtopology => util.stream.Stream.of( | ||
| subtopology.sourceTopics, |
There was a problem hiding this comment.
Can we avoid creating intermediate collections? Either by using Stream.concat, or by just calling anyMatch on each of the collections and combining the boolean results using || for individual collections.
| new StreamsGroupHeartbeatRequestData.Topology() | ||
| .setEpoch(3) | ||
| .setSubtopologies( | ||
| Collections.singletonList(new StreamsGroupHeartbeatRequestData.Subtopology().setSubtopologyId("subtopology") |
There was a problem hiding this comment.
nit: Collections.singletonList -> List.of?
| @@ -10350,10 +10400,32 @@ class KafkaApisTest extends Logging { | |||
| ) | |||
| kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching) | |||
|
|
|||
| val subtotplogy0 = new StreamsGroupDescribeResponseData.Subtopology() | |||
There was a problem hiding this comment.
nit: subtotplogy -> subtopology. Same in other places
| new StreamsGroupDescribeResponseData.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.") |
There was a problem hiding this comment.
"The described group uses topics that the client is not authorized to describe."
This patch filters out the topic describe unauthorized topics from the StreamsGroupHeartbeat and StreamsGroupDescribe response.