KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

FrankYang0529 · 2024-10-16T13:56:01Z

If users subscribe to a topic with invalid name or without permission, they will get some exceptions. Network thread sends MetadataRequest and ConsumerGroupHeartbeatRequest in the background, so there will be some error events in the background queue. When running AsyncKafkaConsumer#unsubscribe, these exceptions should be ignored, or users can't unsubscribe the consumer successfully.

Committer Checklist (excluded from commit message)

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

kirktrue

Is this PR achieving the same effect as PR #17440?

FrankYang0529 · 2024-10-18T04:17:37Z

Yes, it's same. I will close this one.

FrankYang0529 · 2024-10-20T02:05:05Z

Hi @kirktrue, after discussing with @m1a2st, this PR will focus metadata error on unsubscribe and close and his PR #17440 will focus on other functions like beginningOffsets, endOffsets, etc. Thank you.

Could you help me review #17199 first? This PR will rely metric to know there is background event, so we can check unsubscribe and close can work when there is metadata error. Thanks.

m1a2st

Hello @FrankYang0529, Thanks for this PR, left a small suggestion, PTAL

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

lianetm

Hey @FrankYang0529, interesting area, but some high level concerns with the intention here, let me know what you think. Thanks!

lianetm · 2024-10-31T18:08:09Z

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

+    assertDoesNotThrow(new Executable {
+      override def execute(): Unit = consumer.unsubscribe()
+    })


uhm I think we may be not getting the classic consumer behaviour right here. I would expect this assertion passes for the classic consumer here only because of how the test is written (never finds a coordinator), but that doesn't mean that the classic consumer does not throw topic or group auth exceptions on unsubscribe (which is what the changes in this PR are trying to achieve for the new consumer)

I believe classic does not throw GroupAuthException here only because it never finds a coordinator, so the unsubscribes perform no action on the group (but if it had had a known coordinator, the unsubscribe would send a leave group that I expect would fail with GroupAuthException)

I believe classic does not throw TopicAuthException here only because unsubscribe does not poll the network client if it doesn't have a coordinator to send the leave group to.

I could definitely be missing something, but we could validate my expectations with an integration test here:

consumer subscribes to a group successfully (has acls to READ + GROUP + "group-name-from-config")

looses the acls (using removeAndVerifyAcls)

consumer.unsubscribe -> I would expect that this should indeed throw a group authorization exception received in the response to the leave group, or the topic auth exception propagated from metadata (honestly not sure about the precedence, but both should be there)

Hi @lianetm, thanks for the review and suggestion. I try to separate it with 4 test cases:

testConsumeUnsubscribeWithoutTopicPermission

testConsumeCloseWithoutTopicPermission

testConsumeUnsubscribeWithoutGroupPermission

testConsumeCloseWithoutGroupPermission

Although these 4 cases can pass, I need some time to check whether it can really test classic consumer.

sharing more details about my expectations of what the classic consumer does (so the contract we should keep):

do not throw topic or group auth exceptions on unsubscribe simply because it does not wait for the leave group response after sending the request on maybeLeaveGroup

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java

Line 567 in 8cbd2ed

this.coordinator.maybeLeaveGroup("the consumer unsubscribed from all topics");

could throw topic/group auth on close because it does wait for responses (inflight requests) after the maybeLeaveGroup

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java

Line 1140 in 8cbd2ed

if (coordinator != null && !client.awaitPendingRequests(coordinator, timer))

So we need to throw those on close if they exist, but not on unsubscribe (let's align the expectations on the test with this understanding I would say, and we'll validate it). Let me know what you think/find.

Hi @lianetm, thanks for sharing the details in ClassicKafkaConsumer. For ClassicKafkaConsumer#unsubscribe, it doesn't check future object from maybeLeaveGroup, so unsubscribe doesn't throw exceptions. I remain unsubscribe function test for AsyncKafkaConsumer only, and keep close function test for both consumers.

I have another question here. IIUC, we don't need to swallow TopicAuthorizationException, because we don't check topic permission when doing consumer group heartbeat request. So I think we only need to swallow GroupAuthorizationException here. WDYT? Thanks.

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

Lines 3812 to 3815 in 9db5ed0

} else if (!authHelper.authorize(request.context, READ, GROUP, consumerGroupHeartbeatRequest.data.groupId)) {

requestHelper.sendMaybeThrottle(request, consumerGroupHeartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))

CompletableFuture.completedFuture[Unit](())

} else {

For ClassicKafkaConsumer#unsubscribe, it doesn't check future object from maybeLeaveGroup, so unsubscribe doesn't throw exceptions

Agreed, the unused returned value is the reason why the error doesn't bubble up. Still I think we should keep the test for both consumers (it does pass for the classic I expect, this is just the reason why it passes)

I have another question here. IIUC, we don't need to swallow TopicAuthorizationException, because we don't check topic permission when doing consumer group heartbeat request. So I think we only need to swallow GroupAuthorizationException here. WDYT

The trick is that the client internally sends metadata request on poll, and is on those that we could get the topic auth error (that would bubble up in the consumer on processBackgroundEvents). Because of that I think we need to consider that we could receive both (group auth received in a HB response, topic auth received in a metadata response)

lianetm · 2024-10-31T18:09:04Z

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

+    Thread.sleep(3000)
+
+    assertDoesNotThrow(new Executable {
+      override def execute(): Unit = consumer.close()


similar situation with close. I don't believe that the consumer swallows the exceptions, I believe they just don't come out here because there is no known coordinator (so no group request or client poll on close). If my understanding is right, we shouldn't swallow them either on the new consumer

lianetm

Thanks for the updates and responses @FrankYang0529! Some comments mainly to limit the scope to unsubscribe given that there are important changes happening in parallel on consumer.close that would simplify this issue. Let me know what you think

lianetm · 2024-11-11T19:27:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

+            // If users subscribe to a topic with invalid name or without permission, they will get some exceptions.
+            // Because network thread keeps trying to send MetadataRequest or ConsumerGroupHeartbeatRequest in the background,
+            // there will be some error events in the background queue.
+            // When running close, these exceptions should be ignored, or users can't close successfully.
+            processBackgroundEvents(unsubscribeEvent.future(), timer,
+                e -> e instanceof InvalidTopicException || e instanceof GroupAuthorizationException);


what about reverting these changes to keep this PR addressing the unsubscribe only? (aligned with the PR name). I think it will be convenient because there is another ongoing PR #16686 changing the close, and this "releaseAssignmentAndLeaveGroup" does not processBackgroundEvents anymore, so the issue you're trying to solve here disappears (we might remain with the other issue of events unaware of metadata errors, but that is already addressed in another PR #17440 too so let's address those concerns there). Makes sense?

lianetm · 2024-11-11T19:42:24Z

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

+    assertDoesNotThrow(new Executable {
+      override def execute(): Unit = consumer.unsubscribe()
+    })


For ClassicKafkaConsumer#unsubscribe, it doesn't check future object from maybeLeaveGroup, so unsubscribe doesn't throw exceptions

Agreed, the unused returned value is the reason why the error doesn't bubble up. Still I think we should keep the test for both consumers (it does pass for the classic I expect, this is just the reason why it passes)

I have another question here. IIUC, we don't need to swallow TopicAuthorizationException, because we don't check topic permission when doing consumer group heartbeat request. So I think we only need to swallow GroupAuthorizationException here. WDYT

The trick is that the client internally sends metadata request on poll, and is on those that we could get the topic auth error (that would bubble up in the consumer on processBackgroundEvents). Because of that I think we need to consider that we could receive both (group auth received in a HB response, topic auth received in a metadata response)

lianetm · 2024-11-11T19:45:18Z

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

@@ -132,6 +133,84 @@ class GroupAuthorizerIntegrationTest extends BaseRequestTest {
    assertEquals(Set(topic), consumeException.unauthorizedTopics.asScala)
  }

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersConsumerGroupProtocolOnly"))


this test passes for both consumers I expect right? We should enabled it for both

lianetm · 2024-11-11T19:47:27Z

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

+
+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testConsumeCloseWithoutGroupPermission(quorum: String, groupProtocol: String): Unit = {


This test is valuable, but if we agree on leaving this PR for the unsubscribe to leverage the close/callbacks changes in #16686 we should probably bring this test back in a separate PR, after that one goes in. What do you think?

Hi @lianetm, I would like to confirm again: do you mean that we disable close test cases currently, revert change in AsyncKafkaConsumer#releaseAssignmentAndLeaveGroup function, and then we will enable this test after #16686 is merged? Thanks.

Hi @lianetm, thanks for your review. I add test cases for TopicAuthorizationException and disable test cases for close function. Could you help me review again when you have time? Thanks.

It looks like #16686 is almost ready. I will remove @Disabled after it's merged.

lianetm · 2024-11-13T14:15:18Z

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

@@ -132,6 +133,136 @@ class GroupAuthorizerIntegrationTest extends BaseRequestTest {
    assertEquals(Set(topic), consumeException.unauthorizedTopics.asScala)
  }

+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  def testConsumeUnsubscribeWithoutTopicPermission(quorum: String, groupProtocol: String): Unit = {


I wonder if this is better covered at the unit test level only. Here I don't see how we can trust the test is actually testing the unsubscribe changes. The trick is that the topic error comes in a metadata response, but the unsubscribe completes successfully as soon as it gets a response to the HB, so it will always complete ok unless we get a metadata response before the HB response right? It's a really small window btw, because we only processingBackgroundEvents (discover errors) from the moment we send the Unsubscribe (leave HB), to the moment we get a response. Then we stop processing background events, so nothing will be thrown even if it arrives in a response.

The other testConsumeUnsubscribeWithoutGroupPermission makes sense to me, because the group error comes in a HB response, as well as the unsusbcribe response, so we can trust that if the unsubscribe does not throw is because we're indeed swallowing the exception.

Hi @lianetm, thanks for review and suggestion. I move TopicAuthorizationException related test cases to AsyncKafkaConsumerTest. Also, I update the title with close function, so we can add close related test cases in this PR. WDYT? If you prefer to use another PR to include close test cases, I can change it. Thanks.

lianetm

Thanks for the updates @FrankYang0529 ! LGTM.

I would only suggest we update the PR title, to remove the reference to close, because this PR does not change anything related to the close logic (just adds more tests for it), makes sense? Also, in the PR description where it refers to "users can't close ..." I guess it should say unsubscribe, which is what we're fixing in this PR. Thanks!

FrankYang0529 · 2024-11-15T02:34:00Z

Hi @lianetm, thanks for review and suggestion. Updated both title and PR description. Thanks.

…uthorizationException and GroupAuthorizationException Signed-off-by: PoAn Yang <[email protected]>

…uthorizationException and GroupAuthorizationException (apache#17516) Reviewers: Lianet Magrans <[email protected]>, Kirk True <[email protected]>

github-actions bot added core Kafka Broker consumer clients small Small PRs labels Oct 16, 2024

kirktrue mentioned this pull request Oct 17, 2024

KAFKA-17696 New consumer background operations unaware of metadata errors #17440

Merged

3 tasks

kirktrue reviewed Oct 17, 2024

View reviewed changes

FrankYang0529 closed this Oct 18, 2024

FrankYang0529 deleted the KAFKA-17648 branch October 18, 2024 04:17

FrankYang0529 restored the KAFKA-17648 branch October 20, 2024 02:02

FrankYang0529 reopened this Oct 20, 2024

FrankYang0529 marked this pull request as ready for review October 20, 2024 02:15

m1a2st reviewed Oct 24, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java Outdated Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java Outdated Show resolved Hide resolved

FrankYang0529 force-pushed the KAFKA-17648 branch from 9ce814f to 17cb5ec Compare October 25, 2024 12:31

lianetm reviewed Oct 31, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17648 branch from 17cb5ec to bbdf6f0 Compare November 5, 2024 16:03

github-actions bot removed the small Small PRs label Nov 5, 2024

FrankYang0529 force-pushed the KAFKA-17648 branch 2 times, most recently from f7519d2 to 7a3670f Compare November 11, 2024 09:39

FrankYang0529 changed the title ~~KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException~~ KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow GroupAuthorizationException Nov 11, 2024

lianetm reviewed Nov 11, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17648 branch 2 times, most recently from cb6bbe5 to 4b76ef6 Compare November 13, 2024 10:43

FrankYang0529 changed the title ~~KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow GroupAuthorizationException~~ KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException Nov 13, 2024

lianetm reviewed Nov 13, 2024

View reviewed changes

kirktrue added KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels Nov 14, 2024

FrankYang0529 force-pushed the KAFKA-17648 branch 2 times, most recently from 720ec4c to f8880cd Compare November 14, 2024 09:34

FrankYang0529 changed the title ~~KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException~~ KAFKA-17648: AsyncKafkaConsumer#unsubscribe and #close swallow TopicAuthorizationException and GroupAuthorizationException Nov 14, 2024

FrankYang0529 force-pushed the KAFKA-17648 branch 2 times, most recently from 21bf2b5 to ffb000d Compare November 15, 2024 01:26

lianetm approved these changes Nov 15, 2024

View reviewed changes

FrankYang0529 changed the title ~~KAFKA-17648: AsyncKafkaConsumer#unsubscribe and #close swallow TopicAuthorizationException and GroupAuthorizationException~~ KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException Nov 15, 2024

KAFKA-17648: AsyncKafkaConsumer#unsubscribe and #close swallow TopicA…

978d6a4

…uthorizationException and GroupAuthorizationException Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 force-pushed the KAFKA-17648 branch from ffb000d to 978d6a4 Compare November 15, 2024 11:03

lianetm merged commit cc20e78 into apache:trunk Nov 15, 2024
8 checks passed

FrankYang0529 deleted the KAFKA-17648 branch November 15, 2024 14:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

FrankYang0529 commented Oct 16, 2024 •

edited

Loading

kirktrue left a comment •

edited

Loading

FrankYang0529 commented Oct 18, 2024

FrankYang0529 commented Oct 20, 2024

m1a2st left a comment

lianetm left a comment

lianetm Oct 31, 2024

FrankYang0529 Nov 5, 2024

lianetm Nov 6, 2024

FrankYang0529 Nov 11, 2024

lianetm Nov 11, 2024

lianetm Oct 31, 2024

lianetm left a comment

lianetm Nov 11, 2024

lianetm Nov 11, 2024

lianetm Nov 11, 2024

lianetm Nov 11, 2024 •

edited

Loading

FrankYang0529 Nov 13, 2024 •

edited

Loading

FrankYang0529 Nov 13, 2024

FrankYang0529 Nov 13, 2024

lianetm Nov 13, 2024

FrankYang0529 Nov 14, 2024

lianetm left a comment

FrankYang0529 commented Nov 15, 2024

	} else if (!authHelper.authorize(request.context, READ, GROUP, consumerGroupHeartbeatRequest.data.groupId)) {
	requestHelper.sendMaybeThrottle(request, consumerGroupHeartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
	CompletableFuture.completedFuture[Unit](())
	} else {

KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

Conversation

FrankYang0529 commented Oct 16, 2024 • edited Loading

Committer Checklist (excluded from commit message)

kirktrue left a comment • edited Loading

Choose a reason for hiding this comment

FrankYang0529 commented Oct 18, 2024

FrankYang0529 commented Oct 20, 2024

m1a2st left a comment

Choose a reason for hiding this comment

lianetm 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

lianetm 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

lianetm Nov 11, 2024 • edited Loading

Choose a reason for hiding this comment

FrankYang0529 Nov 13, 2024 • edited Loading

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

lianetm left a comment

Choose a reason for hiding this comment

FrankYang0529 commented Nov 15, 2024

FrankYang0529 commented Oct 16, 2024 •

edited

Loading

kirktrue left a comment •

edited

Loading

lianetm Nov 11, 2024 •

edited

Loading

FrankYang0529 Nov 13, 2024 •

edited

Loading