KAFKA-12701: NPE in MetadataRequest when using topic IDs

clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.MetadataRequestData.MetadataRequestTopic;
@@ -92,6 +93,13 @@ public MetadataRequest build(short version) {
             if (!data.allowAutoTopicCreation() && version < 4)
                 throw new UnsupportedVersionException("MetadataRequest versions older than 4 don't support the " +
                         "allowAutoTopicCreation field");
+            if (data.topics() != null) {
+                data.topics().forEach(topic -> {
+                    if (topic.name() == null || topic.topicId() != Uuid.ZERO_UUID)
+                        throw new UnsupportedVersionException("MetadataRequest version " + version  +
+                                " does not support null topic names or non-zero topic IDs.");
+                });
+            }
             return new MetadataRequest(data, version);
         }
 

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

clients/src/test/java/org/apache/kafka/common/requests/MetadataRequestTest.java

@@ -16,16 +16,21 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class MetadataRequestTest {
 
@@ -65,4 +70,24 @@ public void testMetadataRequestVersion() {
         assertEquals(minVersion, builder3.oldestAllowedVersion());
         assertEquals(maxVersion, builder3.latestAllowedVersion());
     }
+
+    @Test
+    public void testTopicIdAndNullTopicNameRequests() {
+        // Construct invalid MetadataRequestTopics. We will build each one separately and ensure the error is thrown.
+        List<MetadataRequestData.MetadataRequestTopic> topics = Arrays.asList(
+                new MetadataRequestData.MetadataRequestTopic().setName(null).setTopicId(Uuid.randomUuid()),
+                new MetadataRequestData.MetadataRequestTopic().setName(null),
+                new MetadataRequestData.MetadataRequestTopic().setTopicId(Uuid.randomUuid()),
+                new MetadataRequestData.MetadataRequestTopic().setName("topic").setTopicId(Uuid.randomUuid()));
+
+        // if version is 10 or 11, the invalid topic metadata should return an error
+        List<Short> invalidVersions = Arrays.asList((short) 10, (short) 11);
+        invalidVersions.forEach(version ->
+            topics.forEach(topic -> {
+                MetadataRequestData metadataRequestData = new MetadataRequestData().setTopics(Collections.singletonList(topic));
+                MetadataRequest.Builder builder = new MetadataRequest.Builder(metadataRequestData);
+                assertThrows(UnsupportedVersionException.class, () -> builder.build(version));
+            })
+        );
+    }
 }

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

@@ -1135,6 +1135,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.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/KafkaApisTest.scala

@@ -39,7 +39,7 @@ import org.apache.kafka.clients.admin.AlterConfigOp.OpType
 import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry}
 import org.apache.kafka.common.acl.AclOperation
 import org.apache.kafka.common.config.ConfigResource
-import org.apache.kafka.common.errors.UnsupportedVersionException
+import org.apache.kafka.common.errors.{InvalidRequestException, UnsupportedVersionException}
 import org.apache.kafka.common.internals.{KafkaFutureImpl, Topic}
 import org.apache.kafka.common.memory.MemoryPool
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
@@ -1052,6 +1052,28 @@ class KafkaApisTest {
         numBrokersNeeded - 1)
   }
 
+  @Test
+  def testInvalidMetadataRequestReturnsError(): Unit = {
+    // Construct invalid MetadataRequestTopics. We will try 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 request = buildRequest(new MetadataRequest(metadataRequestData, version.toShort))
+        val kafkaApis = createKafkaApis()
+
+        assertThrows(classOf[InvalidRequestException], () => kafkaApis.handleTopicMetadataRequest(request))
+      })
+    )
+  }
+
+
   @Test
   def testOffsetCommitWithInvalidPartition(): Unit = {
     val topic = "topic"