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 [email protected]

…ing topic in produce request Signed-off-by: PoAn Yang <[email protected]>

m1a2st

Thank @FrankYang0529 for this patch, some little comments

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

Co-authored-by: Ken Huang <[email protected]>

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

Co-authored-by: Jhen-Yung Hsu <[email protected]>

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

…oduceReqeustVersionLessThan13 Signed-off-by: PoAn Yang <[email protected]>

Signed-off-by: PoAn Yang <[email protected]>

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 <[email protected]>

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.

junrao · 2025-05-06T17:05:18Z

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

+    authorizationForProduceRequestVersionLessThan13(true)
+  }
+
+  def authorizationForProduceRequestVersionLessThan13(createTopic: Boolean): Unit = {


Could this be private?

junrao · 2025-05-06T17:06:24Z

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

+
+  def authorizationForProduceRequestVersionLessThan13(createTopic: Boolean): Unit = {
+    if (createTopic) {
+      sendRequests(mutable.Map(ApiKeys.CREATE_TOPICS -> createTopicsRequest))


Should we use createTopicWithBrokerPrincipal()? This will wait until the topic metadata is propagated to all brokers. Ditto for authorizationFetchV12.

Follow junrao, it would be good to follow testTopicIdAuthorization method?

Use @ParameterizedTest and rename the parameter to withTopicExisting for consistent code style in this file.

Use createTopicWithBrokerPrincipal.

@ParameterizedTest @CsvSource(value = Array("false", "true")) def testTopicIdAuthorization(withTopicExisting: Boolean): Unit = { val topicId = if (withTopicExisting) { createTopicWithBrokerPrincipal(topic) getTopicIds()(topic) } else { Uuid.randomUuid() }

Thanks, I changed to use ParameterizedTest, so we can remove authorizationForProduceRequestVersionLessThan13 and authorizationFetchV12.

Signed-off-by: PoAn Yang <[email protected]>

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

18d0ddc

…ing topic in produce request Signed-off-by: PoAn Yang <[email protected]>

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 <[email protected]>

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

20b5735

Co-authored-by: Ken Huang <[email protected]>

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 <[email protected]>

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 <[email protected]>

revert KafkaApisTest

698f9ac

Signed-off-by: PoAn Yang <[email protected]>

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 <[email protected]>

Merge branch 'trunk' into KAFKA-19234

150e5c9

junrao reviewed May 6, 2025

View reviewed changes

Merge branch 'trunk' into KAFKA-19234

fbcb617

address comment

2368c81

Signed-off-by: PoAn Yang <[email protected]>

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 May 6, 2025

junrao May 6, 2025

Yunyung May 6, 2025

FrankYang0529 May 7, 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

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

FrankYang0529 commented May 4, 2025 •

edited by github-actions bot

Loading

frankvicky left a comment •

edited

Loading