KAFKA-18761: [2/N] List share group offsets with state and auth #19328
Conversation
| if (exception != null) { | ||
| new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup() | ||
| .setGroupId(groupDescribeOffsetsRequest.groupId) | ||
| .setErrorCode(Errors.forException(exception).code) |
There was a problem hiding this comment.
nit: could we create a val Error and use that here
| } | ||
| }); | ||
| return readShareGroupStateSummary(readSummaryRequestData, requestTopicIdToNameMapping, describeShareGroupOffsetsResponseTopicList); | ||
| }); |
There was a problem hiding this comment.
lets add an exceptionally block as well and corresponding unit test
There was a problem hiding this comment.
There is already exception handling inside readShareGroupStateSummary and there is unit testing of it (testDescribeShareGroupAllOffsetsThrowsError for example).
|
Thanks @AndrewJSchofield for the PR. |
| groupDescribeOffsetsRequest: DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup | ||
| ): CompletableFuture[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] = { | ||
| groupCoordinator.describeShareGroupOffsets( | ||
| groupCoordinator.describeShareGroupAllOffsets( | ||
| requestContext, | ||
| groupDescribeOffsetsRequest | ||
| ).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup] { (groupDescribeOffsetsResponse, exception) => | ||
| if (exception != null) { | ||
| new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup() | ||
| .setGroupId(groupDescribeOffsetsRequest.groupId) | ||
| .setErrorCode(Errors.forException(exception).code) |
There was a problem hiding this comment.
nit: could we declare a new val instead of Errors lookup twice?
| @@ -47,9 +46,10 @@ public ListShareGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topi | |||
|
|
|||
| /** | |||
| * Returns the topic partitions whose offsets are to be listed for a share group. | |||
| * {@code null} indicates that offsets of all partitions of the group are to be listed. | |||
| */ | |||
| public Collection<TopicPartition> topicPartitions() { | |||
There was a problem hiding this comment.
What do we do elsewhere in the admin for "all" vs "none"?
There was a problem hiding this comment.
Not sure, but this is precisely the same behaviour as ListConsumerGroupOffsetsSpec. Essentially, if you do new ListShareGroupOffsetsSpec() you get all of them. If you do new ListShareGroupOffsetsSpec().topicPartitions(....) then you get a smaller list. There's no way to ask for none.
| .add(tp.partition()); | ||
| if (spec.topicPartitions() != null) { | ||
| Map<String, List<Integer>> topicPartitionMap = new HashMap<>(); | ||
| spec.topicPartitions().forEach(tp -> topicPartitionMap.computeIfAbsent(tp.topic(), t -> new LinkedList<>()).add(tp.partition())); |
There was a problem hiding this comment.
I realize it's unrelated to this PR, but why are we using a LinkedList here?
There was a problem hiding this comment.
No particular reason I think. I've changed it to ArrayList but it's fairly arbitrary.
This PR approaches completion of Admin.listShareGroupOffsets() and kafka-share-groups.sh --describe --offsets.
Prior to this patch, kafka-share-groups.sh was only able to describe the offsets for partitions which were assigned to active members. Now, the Admin.listShareGroupOffsets() uses the persister's knowledge of the share-partitions which have initialised state. Then, it uses this list to obtain a complete set of offset information.
The PR also implements the topic-based authorisation checking. If Admin.listShareGroupOffsets() is called with a list of topic-partitions specified, the authz checking is performed on the supplied list, returning errors for any topics to which the client is not authorised. If Admin.listShareGroupOffsets() is called without a list of topic-partitions specified, the list of topics is discovered from the persister as described above, and then the response is filtered down to only show the topics to which the client is authorised. This is consistent with other similar RPCs in the Kafka protocol, such as OffsetFetch.