frankvicky commented on code in PR #19515: URL: https://github.com/apache/kafka/pull/19515#discussion_r2059490597
########## core/src/main/scala/kafka/server/KafkaApis.scala: ########## @@ -1043,13 +1045,33 @@ class KafkaApis(val requestChannel: RequestChannel, offsetFetchResponse } else { // Clients are not allowed to see offsets for topics that are not authorized for Describe. - val (authorizedOffsets, _) = authHelper.partitionSeqByAuthorized( + val authorizedNames = authHelper.filterByAuthorized( requestContext, DESCRIBE, TOPIC, offsetFetchResponse.topics.asScala )(_.name) - offsetFetchResponse.setTopics(authorizedOffsets.asJava) + + val topics = new mutable.ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseTopics] + offsetFetchResponse.topics.forEach { topic => + if (authorizedNames.contains(topic.name)) { + if (useTopicIds) { + // If the topic is not provided by the group coordinator, we set it + // using the metadata cache. + if (topic.topicId == Uuid.ZERO_UUID) { + topic.setTopicId(metadataCache.getTopicId(topic.name)) + } + // If we don't have the topic id at all, we skip the topic because + // we can not serialize it without it. Review Comment: I mislook. It will still be `ZERO_UUID` if the corresponding topic ID doesn't exist in the metadata cache. Sorry for the confusing comment. -- 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