[GitHub] [kafka] dajac commented on a diff in pull request #13240: KAFKA-14690: Add topic IDs to OffsetCommit API and propagate for request version >= 9

["dajac (via GitHub)" <gi...@apache.org>]

dajac commented on code in PR #13240:
URL: https://github.com/apache/kafka/pull/13240#discussion_r1105846605


##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -425,35 +425,72 @@ class KafkaApis(val requestChannel: RequestChannel,
       requestHelper.sendMaybeThrottle(request, offsetCommitRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
       CompletableFuture.completedFuture[Unit](())
     } else {
+      val topicNames =
+        if (offsetCommitRequest.version() >= 9)
+          metadataCache.topicIdsToNames()
+        else
+          Collections.emptyMap[Uuid, String]()
+
+      // For version < 9, lookup from topicNames fails and the topic name (which cannot be null) is returned.
+      // For version >= 9, if lookup from topicNames fails, there are two possibilities:
+      //
+      // a) The topic ID was left to default and the topic name should have been populated as a fallback instead.
+      //    If none was provided, null is returned.
+      //
+      // b) The topic ID was not default but is not present in the local topic IDs cache. In this case, because
+      //    clients should make exclusive use of topic name or topic ID, the topic name should be null. If however
+      //    the client provided a topic name, we do not want to use it, because any topic with the same name
+      //    present locally would then have a topic ID which does not match the topic ID in the request.
+      def resolveTopicName(topic: OffsetCommitRequestData.OffsetCommitRequestTopic): String = {
+          val resolvedFromId = topicNames.get(topic.topicId())
+          if (resolvedFromId != null)
+            resolvedFromId
+          else if (offsetCommitRequest.version() < 9 || Uuid.ZERO_UUID.equals(topic.topicId)) {
+            topic.name()
+          } else {
+            null
+          }
+      }
+
       val authorizedTopics = authHelper.filterByAuthorized(
         request.context,
         READ,
         TOPIC,
-        offsetCommitRequest.data.topics.asScala
-      )(_.name)
+        offsetCommitRequest.data.topics.asScala.filter(topic => resolveTopicName(topic) != null)
+      )(resolveTopicName)
 
       val responseBuilder = new OffsetCommitResponse.Builder()
       val authorizedTopicsRequest = new mutable.ArrayBuffer[OffsetCommitRequestData.OffsetCommitRequestTopic]()
       offsetCommitRequest.data.topics.forEach { topic =>
-        if (!authorizedTopics.contains(topic.name)) {
+        val topicName = resolveTopicName(topic)
+        if (topicName == null) {
+          // Topic name cannot be null for version < 9. From version >= 9, topicName is null iff it cannot
+          // be resolved from the local topic IDs cache or topic ID was left to default but no fallback topic
+          // name was provided.
+          responseBuilder.addPartitions[OffsetCommitRequestData.OffsetCommitRequestPartition](

Review Comment:
   There is very likely a bug here. In this case, `topic.name` is `null` and the response builder uses a HashMap keyed by topic name. Therefore, all the topics with an unknown topic id will end up together.



-- 
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