KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18989

dongnuo123 · 2025-02-21T02:12:17Z

This patch filters out the topic describe unauthorized topics from the ConsumerGroupHeartbeat and ConsumerGroupDescribe response.

In ConsumerGroupHeartbeat,

if the request has subscribedTopicNames set, we directly check the authz in KafkaApis and return a topic auth failure in the response if any of the topics is denied.
Otherwise, we check the authz only if a regex refresh is triggered and we do it based on the acl of the consumer that triggered the refresh. If any of the topic is denied, we filter it out from the resolved subscription.

In ConsumerGroupDescribe, we check the authz of the coordinator response. If any of the topic in the group is denied, we remove the described info and add a topic auth failure to the described group. (similar to the group auth failure)

Reviewers: David Jacot [email protected], Lianet Magrans [email protected], Rajini Sivaram [email protected], Chia-Ping Tsai [email protected], TaiJuWu [email protected], TengYao Chi [email protected]

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

dajac

@dongnuo123 Thanks for the patch. I left some high level comments.

core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala

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

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

…onsumer-group-heartbeat-and-describe-feb20

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

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

This reverts commit e80adf4.

…onsumer-group-heartbeat-and-describe-feb20

dajac · 2025-02-22T19:27:18Z

@dongnuo123 Thanks for the update. The patch looks pretty good to me. There are some test failures which are related. Could you please check them with @lianetm?

chia7712

@dongnuo123 thanks for this patch.

chia7712 · 2025-02-23T03:47:38Z

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

@@ -2529,9 +2529,24 @@ class KafkaApis(val requestChannel: RequestChannel,
      requestHelper.sendMaybeThrottle(request, consumerGroupHeartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
      CompletableFuture.completedFuture[Unit](())
    } else {
+      if (consumerGroupHeartbeatRequest.data.subscribedTopicNames != null &&


Does handleShareGroupHeartbeat have similar issue?

Yes but it will be treated separately. This is a critical blocker for 4.0 for the new consumer protocol whereas it is not for share group.

Got it. open https://issues.apache.org/jira/browse/KAFKA-18851 to log it

I think that there is already one open but I cannot find it from my phone.

There it is: https://issues.apache.org/jira/browse/KAFKA-18817

thanks! will close KAFKA-18851 as duplicate

just to complete the story, we need this for KS too, they also have their jira already https://issues.apache.org/jira/browse/KAFKA-18819

chia7712 · 2025-02-23T04:05:01Z

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

+              }
+            }
+          })
+          val authorizedTopics = authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,


It seems handleDescribeGroupsRequest does not require topic authorization. Does it mean users need to update the ACLs when migrating to use AsyncConsumer?

That’s right. The thing is that we cannot apply it on there because the broker is not aware of the topic partitions. It just sees bytes by design.

We could consider parsing the bytes to check the partitions too but as it has been like this since the beginning of the classic protocol. We would need a KIP to discuss it for sure. Personally, I would not change it.

Regarding your second question, the Consumer is not impacted by this as both implementations require topic describe but in different ways. Only the admin client will require extra permissions if not given yet.

@dongnuo123 Could you add a sentence about it in the doc? We could mention this in the ops.html, consumer group section.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

rajinisivaram · 2025-02-24T14:34:49Z

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

@@ -645,6 +645,7 @@ class BrokerServer(
        .withGroupCoordinatorMetrics(new GroupCoordinatorMetrics(KafkaYammerMetrics.defaultRegistry, metrics))
        .withGroupConfigManager(groupConfigManager)
        .withPersister(persister)
+        .withAuthorizer(authorizer.toJava)


Did we consider splitting the processing of heartbeat request in the group coordinator so that topic authorization can continue to be in KafkaApis? Moving some of the authorization out of KafkaApis and into the group coordinator doesn't seem ideal. If we could instead ask the coordinator first for list of topics to filter out for regex and then invoke GroupCoordinatorService.consumerGroupHeartbeat() passing the filtered topics as well, we could avoid splitting out authorization between KafkaApis and group coordinator. And it would avoid knowledge of logIfDenied etc. mentioned in https://github.com/apache/kafka/pull/18989/files#r1966700208 outside of KafkaApis.

if we could instead ask the coordinator first for list of topics to filter out for regex

With the current implementation, the challenge with this is that the resolution and filtering of the regex are done in an async fashion. consumerGroupHeartbeat will return before we finish resolving the list of topics to filter which is done in a separate thread in the coordinator.

But it does make sense that it's a bit strange carrying the authorizer to the coordinator. I think we can work on it as a followup improvement.

Agree, do we have a jira for it @dongnuo123 ?

https://issues.apache.org/jira/browse/KAFKA-18857

@rajinisivaram any concerns with this as follow-up? It will be handled to improve right after this and included in 4.1 as described in KAFKA-18857

lianetm · 2025-02-24T16:36:44Z

I'm taking a closer look at the changes, but regarding the test failures, they show the consumer getting more than one TopicAuth after the ACLs are added (before this change it expected at most 1 more TopicAuth here).

I wonder if that expectation is too strict now, that we send the error via HBs continuously on the interval, even if it fails with error (vs before we would only get the auth error from the Metadata path). The consumer should recover, but I don't think we can guarantee that it will only get 1 TopicAuth after the ACLs are added.

lianetm · 2025-02-24T19:41:27Z

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

+        if (authorizedTopics.size < consumerGroupHeartbeatRequest.data.subscribedTopicNames.size) {
+          val responseData = new ConsumerGroupHeartbeatResponseData()
+            .setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
+            .setErrorMessage("The client is not authorized to describe the provided subscribed topics.")


in this case we had a request with a list of topics, and we're already sending a response with errorCode TOPIC_AUTH_FAILED, so I wonder if adding this message is bringing much value?

Also, it's not consistent with how other APIs handle the same situation. Ex. on the metadata path, we get a request with topic list and if auth fails the response will only include the error code, no custom message. This inconsistency is relevant because it will translate into the consumer btw (whatever topic auth message we get from metadata or HB will bubble up to the consumer.poll). We can always make it consistent on the client side of course, but seems to me that making it consistent here makes more sense given that the message seems redundant. Thoughts?

chia7712 · 2025-02-25T00:24:52Z

@dongnuo123 could you please sync trunk for #18770?

TaiJuWu · 2025-02-25T02:19:07Z

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

+              new ConsumerGroupDescribeResponseData.DescribedGroup()
+                .setGroupId(group.groupId)
+                .setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
+                .setErrorMessage("The group has described topic(s) that the client is not authorized to describe.")


Should this align #18989 (comment) ?

this one I find it valuable, not the same case as my other comment. In this case we don't have topics in the request, so getting a topic_auth in response may not be clear enough, and this generic msg may be helpful.

Got it, thanks.

frankvicky

Thanks for the patch
I have some comments. PTAL

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

…onsumer-group-heartbeat-and-describe-feb20

lianetm

Thanks for the updates @dongnuo123 ! Some minor comments

docs/ops.html

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

lianetm

Thanks for the updates @dongnuo123, LGTM.

Just to recap, we have KAFKA-18857 as follow-up, and related jiras to address this same gap for share (KAFKA-18817) and streams (KAFKA-18819) groups (not 4.0 blockers)

I will wait for the build, and confirmation on this comment https://github.com/apache/kafka/pull/18989/files#r1969914513 in case @rajinisivaram or @chia7712 have any other comments.

Thanks!

lianetm · 2025-02-25T21:16:00Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+                .setMemberId(memberId2)
+                .setMemberEpoch(10)
+                .setRebalanceTimeoutMs(5000)
+                .setSubscribedTopicRegex("foo*|bar*")


just for the record, this makes me notice that we require a HB with regex in this case (and makes sense, because IIRC we only refresh it if it changes or if there is a new topic metadata image, which is not the case here).

I remember there was already a jira/intention to consider time-based refresh, which I expect would help improve the experience in this case. I will find the jira (or file) and update here just for reference.

chia7712

overall LGTM!

chia7712 · 2025-02-26T00:00:43Z

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

+    try {
+      consumeRecords(consumer, numRecords, startingOffset, topic)
+    } catch {
+      case _: TopicAuthorizationException => consumeRecordsIgnoreAuthorizationException(consumer, numRecords, startingOffset, topic)


Is this recursive call necessary? TestUtils.waitUntilTrue already contains a loop, so perhaps we can streamline the check as follows:

TestUtils.waitUntilTrue(() => { try { consumeRecords(consumer) true } catch { case _: TopicAuthorizationException => false } }, "Consumer didn't manage to consume the records within timeout.")

chia7712 · 2025-02-26T06:01:48Z

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

@@ -2592,6 +2607,34 @@ class KafkaApis(val requestChannel: RequestChannel,
            response.groups.addAll(results)
          }

+          // Clients are not allowed to see topics that are not authorized for Describe.
+          val topicsToCheck = response.groups.stream()


one small optimization - could we do this check only if authorizer is defined?

dongnuo123 added 4 commits February 20, 2025 18:33

CGHB

6fe8560

CGHB fix existing UTs

d1bceb4

CGHB with UTs

1fbd09b

describe

01e0675

github-actions bot added triage PRs from the community core Kafka Broker clients labels Feb 21, 2025

dongnuo123 commented Feb 21, 2025

View reviewed changes

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

dongnuo123 mentioned this pull request Feb 21, 2025

KAFKA-18813: [1/N] ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18974

Closed

3 tasks

dongnuo123 added 2 commits February 20, 2025 21:17

spotless check

9d2d286

admin javadoc

6f1dd00

github-actions bot removed the triage PRs from the community label Feb 21, 2025

dajac added KIP-848 The Next Generation of the Consumer Rebalance Protocol Blocker This pull request is identified as solving a blocker for a release. ci-approved labels Feb 21, 2025

dajac reviewed Feb 21, 2025

View reviewed changes

dongnuo123 added 2 commits February 21, 2025 11:24

CGHB

388d11a

CGHB

24f0520

github-actions bot added the KIP-932 Queues for Kafka label Feb 21, 2025

dongnuo123 added 9 commits February 21, 2025 13:58

adapter

4ff3ec6

Merge remote-tracking branch 'ak/trunk' into topic-describe-auth-in-c…

23ddb9e

…onsumer-group-heartbeat-and-describe-feb20

revert adapter change

7c0a042

spotless

808d69d

adapter test

784a13e

spotless check

594e3ab

import control

cd9d2fd

import control

3e11c72

import control

8c6c25c

dajac reviewed Feb 21, 2025

View reviewed changes

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java Outdated Show resolved Hide resolved

dajac reviewed Feb 21, 2025

View reviewed changes

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

dongnuo123 added 4 commits February 22, 2025 11:01

Revert "call authorize for each group"

3ad4bc6

This reverts commit e80adf4.

regex updated log

2b0c043

remove withAuthorizer from CoordinatorShardBuilder

b72ede6

Merge remote-tracking branch 'ak/trunk' into topic-describe-auth-in-c…

768e23e

…onsumer-group-heartbeat-and-describe-feb20

chia7712 reviewed Feb 23, 2025

View reviewed changes

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java Show resolved Hide resolved

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java Outdated Show resolved Hide resolved

rajinisivaram reviewed Feb 24, 2025

View reviewed changes

dongnuo123 added 2 commits February 24, 2025 11:25

don't log if denied when filtering regex

6f6c978

ops.html

ecf04ee

dongnuo123 added 3 commits February 24, 2025 12:29

testNoConsumeWithoutDescribeAclViaSubscribe

71cc9fe

testNoConsumeWithoutDescribeAclViaSubscribe

62e5384

testNoConsumeWithoutDescribeAclViaSubscribe

82b40da

lianetm reviewed Feb 24, 2025

View reviewed changes

dongnuo123 added 2 commits February 24, 2025 15:18

AuthorizerIntegrationTest

00654ac

remove error description from the response

1dcec7f

dongnuo123 changed the title ~~KAFKA-18813: [1/N] ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe~~ KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe Feb 24, 2025

add error message to describe response

6430230

TaiJuWu reviewed Feb 25, 2025

View reviewed changes

frankvicky reviewed Feb 25, 2025

View reviewed changes

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

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

dongnuo123 added 2 commits February 25, 2025 09:34

Merge remote-tracking branch 'ak/trunk' into topic-describe-auth-in-c…

5b64632

…onsumer-group-heartbeat-and-describe-feb20

refactor filtering describe response

0b93611

lianetm reviewed Feb 25, 2025

View reviewed changes

docs/ops.html Outdated Show resolved Hide resolved

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java Show resolved Hide resolved

UTs

46bcca9

lianetm approved these changes Feb 25, 2025

View reviewed changes

chia7712 approved these changes Feb 26, 2025

View reviewed changes

frankvicky approved these changes Feb 26, 2025

View reviewed changes

chia7712 reviewed Feb 26, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18989

KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18989

dongnuo123 commented Feb 21, 2025 •

edited by lianetm

Loading

dajac left a comment

dajac commented Feb 22, 2025

chia7712 left a comment

chia7712 Feb 23, 2025

dajac Feb 23, 2025

chia7712 Feb 23, 2025

dajac Feb 23, 2025

dajac Feb 23, 2025

chia7712 Feb 23, 2025

lianetm Feb 24, 2025

chia7712 Feb 23, 2025

dajac Feb 23, 2025

rajinisivaram Feb 24, 2025

dongnuo123 Feb 24, 2025 •

edited

Loading

lianetm Feb 24, 2025

dongnuo123 Feb 24, 2025

lianetm Feb 25, 2025

lianetm commented Feb 24, 2025 •

edited

Loading

lianetm Feb 24, 2025

chia7712 commented Feb 25, 2025

TaiJuWu Feb 25, 2025

lianetm Feb 25, 2025

TaiJuWu Feb 25, 2025

frankvicky left a comment

lianetm left a comment

lianetm left a comment •

edited

Loading

lianetm Feb 25, 2025

chia7712 left a comment

chia7712 Feb 26, 2025

chia7712 Feb 26, 2025

KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18989

Are you sure you want to change the base?

KAFKA-18813: ConsumerGroupHeartbeat API and ConsumerGroupDescribe API must check topic describe #18989

Conversation

dongnuo123 commented Feb 21, 2025 • edited by lianetm Loading

dajac left a comment

Choose a reason for hiding this comment

dajac commented Feb 22, 2025

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

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

dongnuo123 Feb 24, 2025 • 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

lianetm commented Feb 24, 2025 • edited Loading

Choose a reason for hiding this comment

chia7712 commented Feb 25, 2025

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

Choose a reason for hiding this comment

lianetm left a comment

Choose a reason for hiding this comment

lianetm left a comment • edited Loading

Choose a reason for hiding this 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

dongnuo123 commented Feb 21, 2025 •

edited by lianetm

Loading

dongnuo123 Feb 24, 2025 •

edited

Loading

lianetm commented Feb 24, 2025 •

edited

Loading

lianetm left a comment •

edited

Loading