dajac commented on code in PR #19461:
URL: https://github.com/apache/kafka/pull/19461#discussion_r2047353585
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -271,40 +271,62 @@ class KafkaApis(val requestChannel: RequestChannel,
): CompletableFuture[Unit] = {
val offsetCommitRequest = request.body[OffsetCommitRequest]
- // Reject the request if not authorized to the group
+ // Reject the request if not authorized to the group.
if (!authHelper.authorize(request.context, READ, GROUP,
offsetCommitRequest.data.groupId)) {
requestHelper.sendMaybeThrottle(request,
offsetCommitRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
CompletableFuture.completedFuture[Unit](())
} else {
+ val useTopicIds =
OffsetCommitResponse.useTopicIds(request.header.apiVersion)
+
+ if (useTopicIds) {
+ offsetCommitRequest.data.topics.forEach { topic =>
+ if (topic.topicId != Uuid.ZERO_UUID) {
+ metadataCache.getTopicName(topic.topicId).ifPresent(name =>
topic.setName(name))
+ }
+ }
+ }
+
val authorizedTopics = authHelper.filterByAuthorized(
request.context,
READ,
TOPIC,
offsetCommitRequest.data.topics.asScala
)(_.name)
- val responseBuilder = new OffsetCommitResponse.Builder()
+ val responseBuilder = OffsetCommitResponse.newBuilder(useTopicIds)
val authorizedTopicsRequest = new
mutable.ArrayBuffer[OffsetCommitRequestData.OffsetCommitRequestTopic]()
offsetCommitRequest.data.topics.forEach { topic =>
- if (!authorizedTopics.contains(topic.name)) {
+ if (useTopicIds && topic.name.isEmpty) {
Review Comment:
If the topic name is empty here, it means that we were not able to resolve
the topic id at L284. Therefore, we need to reject the topic with unknown topic
id. We only need to do this if topic ids are used.
--
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
