KAFKA-19234: broker should return UNAUTHORIZATION error for non-existing topic in produce request #19635

FrankYang0529 · 2025-05-04T13:14:37Z

Since topic name is sensitive information, it should return a
TOPIC_AUTHORIZATION_FAILED error for non-existing topic. The Fetch
request also follows this pattern.

Co-authored-by: John Doe zh2725284321@gmail.com

…ing topic in produce request Signed-off-by: PoAn Yang <payang@apache.org>

m1a2st

Thank @FrankYang0529 for this patch, some little comments

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

Co-authored-by: Ken Huang <s7133700@gmail.com>

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

Co-authored-by: Jhen-Yung Hsu <yungyung7654321@gmail.com>

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

…oduceReqeustVersionLessThan13 Signed-off-by: PoAn Yang <payang@apache.org>

Signed-off-by: PoAn Yang <payang@apache.org>

TaiJuWu

LGTM.

chia7712 · 2025-05-05T13:48:33Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -924,6 +924,33 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
    sendRequests(requestKeyToRequest, false, topicNames)
  }

+  @Test
+  def testAuthorizationWithTopicNotExistingForProduceReqeustVersionLessThan13(): Unit = {


@Test def testAuthorizationWithTopicNotExistingForProduceRequestVersionLessThan13(): Unit = { sendRequests(mutable.Map(ApiKeys.CREATE_TOPICS -> createTopicsRequest)) def createTpd(version: Int) = if (version <= 12) new ProduceRequestData.TopicProduceData() .setName(tp.topic()) else new ProduceRequestData.TopicProduceData() .setTopicId(getTopicIds()(tp.topic())) for (version <- ApiKeys.PRODUCE.oldestVersion to ApiKeys.PRODUCE.latestVersion) { val request = requests.ProduceRequest.builder(new ProduceRequestData() .setTopicData(new ProduceRequestData.TopicProduceDataCollection( util.List.of(createTpd(version) .setPartitionData(util.List.of( new ProduceRequestData.PartitionProduceData() .setIndex(tp.partition) .setRecords(MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("test".getBytes)))))) .iterator)) .setAcks(1.toShort) .setTimeoutMs(5000)) .build(version.toShort) val data = connectAndReceive[AbstractResponse](request, listenerName = listenerName).asInstanceOf[ProduceResponse].data().responses() assertEquals(1, data.size()) val response = if (version <= 12) data.find(tp.topic(), Uuid.ZERO_UUID) else data.find("", getTopicIds()(tp.topic())) assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED.code(), response.partitionResponses().asScala.find(_.index == part).get.errorCode, s"unexpected error for produce request version $version") } }

we need to test version 13, but it should use topic id instead of topic name.

Hi @chia7712, the version 13 is covered by testAuthorizationWithTopicNotExisting. This PR focuses on old versions behavior.

Could you please add comments for the version=13. Additionally, there is a typo: Reqeust

junrao

@FrankYang0529 : Thanks for the PR. Left a comment.

junrao · 2025-05-05T17:03:53Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+      val request = requests.ProduceRequest.builder(new ProduceRequestData()
+          .setTopicData(new ProduceRequestData.TopicProduceDataCollection(
+            util.List.of(new ProduceRequestData.TopicProduceData()
+                .setName(tp.topic())


This topic exists on the broker, right?

We expect to return TOPIC_AUTHORIZATION_FAILED (if it failed to authorize) regardless of topic existence if the request is using topic name. Maybe we should test both cases - 1) topic exists and 2) topic does not exist (similar to testAuthorizationFetchV12WithTopicNotExisting)

We could. I am just pointing out this doesn't match the test name, which says WithTopicNotExisting.

Hi @junrao, This case follows testAuthorizationWithTopicNotExisting. In AuthorizerIntegrationTest, if a case needs a existing topic like case testAuthorizationWithTopicExisting, it needs to call sendRequests(mutable.Map(ApiKeys.CREATE_TOPICS -> createTopicsRequest)).

@FrankYang0529 could you please add the test case for the existent topic? Additionally, it would be useful to add case of existent topic for fetch RPC too.

junrao · 2025-05-05T17:54:19Z

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

@@ -413,8 +413,6 @@ class KafkaApis(val requestChannel: RequestChannel,
        val topicPartition = new TopicPartition(topicName, partition.index())
        if (topicName.isEmpty)


This is another subtle existing issue. If a produce request before v13 has an empty topic name, we used to return UNKNOWN_TOPIC_OR_PARTITION. Now, we return UNKNOWN_TOPIC_ID. It's probably better to change the condition to if (topicName.isEmpty && topic.topicId().equals(Uuid.ZERO_UUID)).

It's probably better to change the condition to if (topicName.isEmpty && topic.topicId().equals(Uuid.ZERO_UUID)).

The default value of topic.topicId() is Uuid.ZERO_UUID, and hence if (topicName.isEmpty && topic.topicId().equals(Uuid.ZERO_UUID))is always true if request is before v13. Maybe we can try another condition:if (topicName.isEmpty && version > 12`?

Sorry, I meant if (topicName.isEmpty && !topic.topicId().equals(Uuid.ZERO_UUID)).

junrao · 2025-05-05T17:55:35Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+      val request = requests.ProduceRequest.builder(new ProduceRequestData()
+          .setTopicData(new ProduceRequestData.TopicProduceDataCollection(
+            util.List.of(new ProduceRequestData.TopicProduceData()
+                .setName(tp.topic())


We could. I am just pointing out this doesn't match the test name, which says WithTopicNotExisting.

frankvicky

Overall, LGTM assuming the remaining comments are addressed.

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 · 2025-05-06T12:50:59Z

@junrao @chia7712 Thanks for review. I addressed all comments. Could you take a look when you have time? Thanks.

junrao

@FrankYang0529 : Thanks for the updated PR. A couple of more comments.

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

Signed-off-by: PoAn Yang <payang@apache.org>

junrao

@FrankYang0529 : Thanks for the updated PR. A couple of more comments.

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

Signed-off-by: PoAn Yang <payang@apache.org>

chia7712

@FrankYang0529 thanks for updates.

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

chia7712 · 2025-05-08T06:12:09Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

   */
  @ParameterizedTest
-  @ValueSource(strings = Array("kraft"))
-  def testAuthorizationFetchV12WithTopicNotExisting(quorum: String): Unit = {
+  @CsvSource(value = Array("false", "true"))


could you please use @ValueSource(booleans = Array(true, false)) instead?

Updated it. Thanks.

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

…m13ToNewest Signed-off-by: PoAn Yang <payang@apache.org>

junrao

@FrankYang0529 : Thanks for the updated PR. A few more comments.

junrao · 2025-05-08T17:56:46Z

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

@@ -411,10 +411,9 @@ class KafkaApis(val requestChannel: RequestChannel,
        }

        val topicPartition = new TopicPartition(topicName, partition.index())
-        if (topicName.isEmpty)
+        // To compatible with the old version, only return UNKNOWN_TOPIC_ID if topicId is not default value


To compatible => To be compatible

junrao · 2025-05-08T18:10:26Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala


    val requestKeyToRequest = mutable.LinkedHashMap[ApiKeys, AbstractRequest](
      ApiKeys.METADATA -> createMetadataRequest(allowAutoTopicCreation = true),
-      ApiKeys.PRODUCE -> createProduceRequest,
+      ApiKeys.PRODUCE -> createProduceRequest("", getTopicIds().getOrElse(tp.topic, Uuid.ZERO_UUID), ApiKeys.PRODUCE.latestVersion()),


Since we expect the topicId to be present, should we throw an exception if the topicId can't be found?

junrao · 2025-05-08T18:46:41Z

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

@@ -411,10 +411,9 @@ class KafkaApis(val requestChannel: RequestChannel,
        }

        val topicPartition = new TopicPartition(topicName, partition.index())
-        if (topicName.isEmpty)
+        // To compatible with the old version, only return UNKNOWN_TOPIC_ID if topicId is not default value
+        if (topicName.isEmpty && !topic.topicId().equals(Uuid.ZERO_UUID))


If the request version is 13 and the topicId is set to 0, we should return UNKNOWN_TOPIC_ID to match what's in the fetch request. Only if the request version is below 13 and topicName is empty, should we return UNKNOWN_TOPIC_OR_PARTITION. So, it seems that we should do fetchRequest.version() >= 13 instead of !topic.topicId().equals(Uuid.ZERO_UUID).

That makes sense. I left the similar comment before - #19635 (comment)

junrao · 2025-05-08T18:49:01Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+   * and topic name is used in the request. Even if the topic doesn't exist, we return TOPIC_AUTHORIZATION_FAILED to
+   * prevent leaking the topic name.
+   * This case covers produce request version from oldest to 12.
+   * The newer version is covered by testAuthorizationWithTopicNotExisting.


testAuthorizationWithTopicNotExisting => testAuthorizationWithTopicNotExisting and testAuthorizationWithTopicExisting

junrao · 2025-05-08T18:49:16Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+   * and topic name is used in the request. Even if the topic doesn't exist, we return TOPIC_AUTHORIZATION_FAILED to
+   * prevent leaking the topic name.
+   * This case covers fetch request version from oldest to 12.
+   * The newer version is covered by testAuthorizationWithTopicNotExisting.


testAuthorizationWithTopicNotExisting => testAuthorizationWithTopicNotExisting and testAuthorizationWithTopicExisting

junrao · 2025-05-08T18:50:31Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -1021,7 +1030,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
   * The newer version is covered by testAuthorizationWithTopicNotExisting.
   */
  @ParameterizedTest
-  @CsvSource(value = Array("false", "true"))
+  @ValueSource(booleans = Array(true, false))


Should we change the ordering in testTopicIdAuthorization too to be consistent??

Signed-off-by: PoAn Yang <payang@apache.org>

junrao

@FrankYang0529 : Thanks for the updated PR. A couple of more comments.

junrao · 2025-05-09T23:30:48Z

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

@@ -411,10 +411,9 @@ class KafkaApis(val requestChannel: RequestChannel,
        }

        val topicPartition = new TopicPartition(topicName, partition.index())
-        if (topicName.isEmpty)
+        // To be compatible with the old version, only return UNKNOWN_TOPIC_ID if topic name is not set and request version is bigger than 12


if topic name is not set and request version is bigger than 12 => if request version uses topicId, but the corresponding topic name can't be found

junrao · 2025-05-10T00:08:11Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+   * The produce request only supports topic id above version 13.
+   */
+  @Test
+  def testEmptyTopicNameAndIDForProduceVersionFrom13ToNewest(): Unit = {


This test is covered by testAuthorizationWithTopicNotExisting, right?

KAFKA-19234: broker should return UNAUTHORIZATION error for non-exist…

18d0ddc

…ing topic in produce request Signed-off-by: PoAn Yang <payang@apache.org>

github-actions bot added core Kafka Broker small Small PRs labels May 4, 2025

FrankYang0529 mentioned this pull request May 4, 2025

KAFKA-19234: broker should return UNAUTHORIZATION error for non-exist… #19632

Closed

m1a2st reviewed May 4, 2025

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Outdated Show resolved Hide resolved

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Outdated Show resolved Hide resolved

FrankYang0529 and others added 2 commits May 4, 2025 21:26

Update core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

24c300f

Co-authored-by: Ken Huang <s7133700@gmail.com>

Update core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

20b5735

Co-authored-by: Ken Huang <s7133700@gmail.com>

FrankYang0529 requested a review from m1a2st May 4, 2025 13:26

FrankYang0529 added the ci-approved label May 4, 2025

Yunyung reviewed May 4, 2025

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Outdated Show resolved Hide resolved

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

62f136a

Co-authored-by: Jhen-Yung Hsu <yungyung7654321@gmail.com>

FrankYang0529 requested review from Yunyung, chia7712 and frankvicky May 4, 2025 16:01

chia7712 reviewed May 5, 2025

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Outdated Show resolved Hide resolved

FrankYang0529 added 3 commits May 5, 2025 14:53

Merge branch 'trunk' into KAFKA-19234

b1dacbb

KAFKA-19234: Add test case testAuthorizationWithTopicNotExistingForPr…

1df0668

…oduceReqeustVersionLessThan13 Signed-off-by: PoAn Yang <payang@apache.org>

revert KafkaApisTest

698f9ac

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 requested a review from chia7712 May 5, 2025 12:53

TaiJuWu approved these changes May 5, 2025

View reviewed changes

chia7712 reviewed May 5, 2025

View reviewed changes

chia7712 approved these changes May 5, 2025

View reviewed changes

junrao reviewed May 5, 2025

View reviewed changes

frankvicky approved these changes May 6, 2025

View reviewed changes

FrankYang0529 added 2 commits May 6, 2025 18:38

address comment

8c845f4

Signed-off-by: PoAn Yang <payang@apache.org>

Merge branch 'trunk' into KAFKA-19234

150e5c9

junrao reviewed May 6, 2025

View reviewed changes

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala Outdated Show resolved Hide resolved

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala Outdated Show resolved Hide resolved

Merge branch 'trunk' into KAFKA-19234

fbcb617

address comment

2368c81

Signed-off-by: PoAn Yang <payang@apache.org>

junrao reviewed May 7, 2025

View reviewed changes

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala Outdated Show resolved Hide resolved

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala Outdated Show resolved Hide resolved

FrankYang0529 added 3 commits May 8, 2025 09:33

Merge branch 'trunk' into KAFKA-19234

d35d552

address comment

159d3ec

Signed-off-by: PoAn Yang <payang@apache.org>

Merge branch 'trunk' into KAFKA-19234

4cd03ea

chia7712 reviewed May 8, 2025

View reviewed changes

add comment and test case testEmptyTopicNameAndIDForProduceVersionFro…

758663d

…m13ToNewest Signed-off-by: PoAn Yang <payang@apache.org>

github-actions bot removed the small Small PRs label May 8, 2025

junrao reviewed May 8, 2025

View reviewed changes

FrankYang0529 added 2 commits May 9, 2025 15:51

Merge branch 'trunk' into KAFKA-19234

73e5756

address comment

70067da

Signed-off-by: PoAn Yang <payang@apache.org>

junrao reviewed May 10, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-19234: broker should return UNAUTHORIZATION error for non-existing topic in produce request #19635

KAFKA-19234: broker should return UNAUTHORIZATION error for non-existing topic in produce request #19635

FrankYang0529 commented May 4, 2025 •

edited by github-actions bot

Loading

m1a2st left a comment

TaiJuWu left a comment

chia7712 May 5, 2025

FrankYang0529 May 5, 2025

chia7712 May 5, 2025

junrao left a comment

junrao May 5, 2025

chia7712 May 5, 2025

junrao May 5, 2025

FrankYang0529 May 6, 2025

chia7712 May 6, 2025

junrao May 5, 2025

chia7712 May 5, 2025

junrao May 5, 2025

junrao May 5, 2025

frankvicky left a comment •

edited

Loading

FrankYang0529 commented May 6, 2025

junrao left a comment

junrao left a comment

chia7712 left a comment

chia7712 May 8, 2025

FrankYang0529 May 8, 2025

junrao left a comment

junrao May 8, 2025

junrao May 8, 2025

junrao May 8, 2025

chia7712 May 9, 2025

junrao May 8, 2025

junrao May 8, 2025

junrao May 8, 2025

junrao left a comment

junrao May 9, 2025

junrao May 10, 2025

		@@ -413,8 +413,6 @@ class KafkaApis(val requestChannel: RequestChannel,
		val topicPartition = new TopicPartition(topicName, partition.index())
		if (topicName.isEmpty)

KAFKA-19234: broker should return UNAUTHORIZATION error for non-existing topic in produce request #19635

Are you sure you want to change the base?

KAFKA-19234: broker should return UNAUTHORIZATION error for non-existing topic in produce request #19635

Conversation

FrankYang0529 commented May 4, 2025 • edited by github-actions bot Loading

m1a2st left a comment

Choose a reason for hiding this comment

TaiJuWu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankvicky left a comment • edited Loading

Choose a reason for hiding this comment

FrankYang0529 commented May 6, 2025

junrao left a comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented May 4, 2025 •

edited by github-actions bot

Loading

frankvicky left a comment •

edited

Loading