jolshan commented on code in PR #14310:
URL: https://github.com/apache/kafka/pull/14310#discussion_r1310545133
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1452,62 +1452,63 @@ class KafkaApis(val requestChannel: RequestChannel,
private def fetchAllOffsetsForGroup(
requestContext: RequestContext,
- groupOffsetFetch: OffsetFetchRequestData.OffsetFetchRequestGroup,
+ offsetFetchRequest: OffsetFetchRequestData.OffsetFetchRequestGroup,
requireStable: Boolean
): CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup] = {
groupCoordinator.fetchAllOffsets(
requestContext,
- groupOffsetFetch.groupId,
+ offsetFetchRequest,
requireStable
- ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] { (offsets,
exception) =>
+ ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] {
(offsetFetchResponse, exception) =>
if (exception != null) {
new OffsetFetchResponseData.OffsetFetchResponseGroup()
- .setGroupId(groupOffsetFetch.groupId)
+ .setGroupId(offsetFetchRequest.groupId)
.setErrorCode(Errors.forException(exception).code)
} else {
// Clients are not allowed to see offsets for topics that are not
authorized for Describe.
val (authorizedOffsets, _) = authHelper.partitionSeqByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
- offsets.asScala
+ offsetFetchResponse.topics.asScala
)(_.name)
new OffsetFetchResponseData.OffsetFetchResponseGroup()
- .setGroupId(groupOffsetFetch.groupId)
+ .setGroupId(offsetFetchRequest.groupId)
.setTopics(authorizedOffsets.asJava)
}
}
}
private def fetchOffsetsForGroup(
requestContext: RequestContext,
- groupOffsetFetch: OffsetFetchRequestData.OffsetFetchRequestGroup,
+ offsetFetchRequest: OffsetFetchRequestData.OffsetFetchRequestGroup,
requireStable: Boolean
): CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup] = {
// Clients are not allowed to see offsets for topics that are not
authorized for Describe.
val (authorizedTopics, unauthorizedTopics) =
authHelper.partitionSeqByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
- groupOffsetFetch.topics.asScala
+ offsetFetchRequest.topics.asScala
)(_.name)
groupCoordinator.fetchOffsets(
requestContext,
- groupOffsetFetch.groupId,
- authorizedTopics.asJava,
+ new OffsetFetchRequestData.OffsetFetchRequestGroup()
+ .setGroupId(offsetFetchRequest.groupId)
+ .setTopics(authorizedTopics.asJava),
requireStable
- ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] {
(topicOffsets, exception) =>
+ ).handle[OffsetFetchResponseData.OffsetFetchResponseGroup] {
(offsetFetchResponse, exception) =>
if (exception != null) {
new OffsetFetchResponseData.OffsetFetchResponseGroup()
- .setGroupId(groupOffsetFetch.groupId)
+ .setGroupId(offsetFetchRequest.groupId)
.setErrorCode(Errors.forException(exception).code)
} else {
val response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
Review Comment:
couldn't we just use the response returned here? Or is it a problem because
we can't add the unauthorized ones. It is a bit unfortunate we have to create
the object twice. (Here and in fetching all offsets)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
