KAFKA-12701: NPE in MetadataRequest when using topic IDs

clients/src/main/resources/common/message/MetadataRequest.json

@@ -33,7 +33,8 @@
     //
     // Version 9 is the first flexible version.
     //
-    // Version 10 adds topicId.
+    // Version 10 adds topicId and allows name field to be null. However, this functionality was not implemented on the server.
+    // Versions 10 and 11 should not use the topicId field or set topic name to null.
     //
     // Version 11 deprecates IncludeClusterAuthorizedOperations field. This is now exposed
     // by the DescribeCluster API (KIP-700).

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -1142,6 +1142,19 @@ class KafkaApis(val requestChannel: RequestChannel,
     val metadataRequest = request.body[MetadataRequest]
     val requestVersion = request.header.apiVersion
 
+    // Topic IDs are not supported for versions 10 and 11. Topic names can not be null in these versions.
+    if (metadataRequest.version() >= 10 && !metadataRequest.isAllTopics) {
+      metadataRequest.data().topics().forEach{ topic =>
+        // If null, set to the empty string, since the response does not allow null.
+        if (topic.name() == null) {
+          topic.setName("")
+          throw new InvalidRequestException(s"Topic name can not be null for version ${metadataRequest.version()}")
+        } else if (topic.topicId() != Uuid.ZERO_UUID) {
+          throw new InvalidRequestException(s"Topic IDs are not supported in requests for version ${metadataRequest.version()}")
+        }
+      }
+    }
+
     val topics = if (metadataRequest.isAllTopics)
       metadataCache.getAllTopics()
     else

core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala

@@ -17,12 +17,13 @@
 
 package kafka.server
 
-import java.util.Optional
+import java.util.{Collections, Optional}
 
 import kafka.utils.TestUtils
 import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.errors.UnsupportedVersionException
 import org.apache.kafka.common.internals.Topic
+import org.apache.kafka.common.message.MetadataRequestData
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.{MetadataRequest, MetadataResponse}
 import org.apache.kafka.metadata.BrokerState
@@ -234,6 +235,32 @@ class MetadataRequestTest extends AbstractMetadataRequestTest {
     }
   }
 
+  @Test
+  def testInvalidMetadataRequestReturnsError(): Unit = {
+    val replicaAssignment = Map(0 -> Seq(1, 2, 0))
+    val topic1 = "topic1"
+    val topic2 = "topic2"
+    createTopic(topic1, replicaAssignment)
+    createTopic(topic2, replicaAssignment)
+
+    // Construct invalid MetadataRequestTopics. We will send each one separately and ensure the error is thrown.
+    val topics = List(new MetadataRequestData.MetadataRequestTopic().setName(null).setTopicId(Uuid.randomUuid()),
+                      new MetadataRequestData.MetadataRequestTopic().setName(null),
+                      new MetadataRequestData.MetadataRequestTopic().setTopicId(Uuid.randomUuid()),
+                      new MetadataRequestData.MetadataRequestTopic().setName(topic1).setTopicId(Uuid.randomUuid()))
+
+    // if version is 10 or 11, the invalid topic metadata should return an error
+    val invalidVersions = Set(10, 11)
+    invalidVersions.foreach( version =>
+      topics.foreach(topic => {
+        val metadataRequestData = new MetadataRequestData().setTopics(Collections.singletonList(topic))
+        val resp = sendMetadataRequest(new MetadataRequest(metadataRequestData, version.toShort), Some(controllerSocketServer))
+        assertEquals(1, resp.topicMetadata.size)
+        assertEquals(1, resp.errorCounts().get(Errors.INVALID_REQUEST))
+      })
+    )
+  }
+
   /**
     * Preferred replica should be the first item in the replicas list
     */