KAFKA-12701: NPE in MetadataRequest when using topic IDs

[jolshan]

We prevent handling MetadataRequests where the topic name is null (to prevent NPE) as well as prevent requests that set topic IDs since this functionality has not yet been implemented. When we do implement in in #9769, we should bump the request/response version.

Should also cherry-pick these changes to 2.8 for the next release.

Added tests to ensure the error is thrown.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[jolshan]

Test failures look unrelated. Some of the usual suspects like RaftClusterTest.testCreateClusterAndCreateAndManyTopicsWithManyPartitions() and MirrorConnectorsIntegrationSSLTest/MirrorConnectorsIntegrationTesttests

[dengziming]

I think this makes sense

[ijuma]

Thanks for the PRs, left a few comments.

[ijuma]

LGTM, a minor nit below.

[ijuma]

I think you need to merge master to this branch or rebase this branch on master. Jenkins is failing due to:

[2021-06-10T21:42:48.226Z] [Error] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-10584/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:1082: not enough arguments for method handle: (request: kafka.network.RequestChannel.Request, requestLocal: kafka.server.RequestLocal)Unit.

[2021-06-10T21:42:48.226Z] Unspecified value parameter requestLocal.

[jolshan]

Ok. This should fix the build. But now it will be a little bit extra work to cherry-pick to 2.8.x 😅

[dengziming]

Ok. This should fix the build. But now it will be a little bit extra work to cherry-pick to 2.8.x 😅

@jolshan you can open another PR target for 2.8.x if it's difficult to cherry-pick 😂

[ijuma]

Thanks, a couple more comments below.

[ijuma]

Thanks, LGTM. Just one nit question below.

[jolshan]

Ah I see I didn't need that extra declaration but you also removed the comment. Not a huge deal though.

[ijuma]

Yeah, that unintentional while using the GitHub editor. Will add it back.

[ijuma]

LGTM, thanks!

[ijuma]

@jolshan The test needs to be adjusted for 2.8. Can you please look into that? I started on it, but it also needs to handle the fact that an exception is thrown from the handleTopicMetadata method:

  @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 metadataRequest = new MetadataRequest(metadataRequestData, version.toShort)
        val request = buildRequest(metadataRequest)
        val capturedResponse = expectNoThrottling()
        EasyMock.replay(replicaManager, clientRequestQuotaManager,
          autoTopicCreationManager, forwardingManager, clientControllerQuotaManager, groupCoordinator, txnCoordinator)
        createKafkaApis().handleTopicMetadataRequest(request)

        val response = readResponse(metadataRequest, capturedResponse)
          .asInstanceOf[MetadataResponse]
        assertEquals(1, response.topicMetadata.size)
        assertEquals(1, response.errorCounts.get(Errors.INVALID_REQUEST))
        response.data.topics.forEach(topic => assertNotEquals(null, topic.name))
        reset(requestChannel)
      })
    )
  }

[jolshan]

I didn't see a branch or PR with the 2.8 version, so I opened my own here: #10885