MINOR: Simplify OffsetFetchResponse #19642
Conversation
…set-fetch-response
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
Outdated
Show resolved
Hide resolved
.../src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
Outdated
Show resolved
Hide resolved
| topic.partitions().forEach(partition -> { | ||
| OffsetFetchResponsePartition newPartition; | ||
|
|
||
| if (version < 2 && group.errorCode() != Errors.NONE.code()) { |
There was a problem hiding this comment.
There is a bug here. This logic assumes that partitions are given along side a group level error. However, looking at KafkaApis, it is not the case. Hence, a group level error is not returned at all for version 1 of the API. I will file a Jira and address it separately.
There was a problem hiding this comment.
…set-fetch-response
| Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.TOPIC_AUTHORIZATION_FAILED); | ||
| Errors.UNKNOWN_TOPIC_OR_PARTITION, | ||
| Errors.TOPIC_AUTHORIZATION_FAILED, | ||
| Errors.UNSTABLE_OFFSET_COMMIT |
There was a problem hiding this comment.
Is there a reason why we need this error here? this one is expected only if v>7 (requires stable), but PARTITION_ERRORS is used only if v<2 (when we want to bubble up errors)
There was a problem hiding this comment.
Not really. I was debating whether we should be exhaustive here and list all the partition errors or not. Let me remove it and add a comment.
| Map<String, Errors> errorMap, | ||
| Map<String, Map<TopicPartition, PartitionData>> responseData | ||
| ) { | ||
| return new OffsetFetchResponse(throttleMs, errorMap, responseData); |
There was a problem hiding this comment.
throttleMs doesn't seem needed after this (on ln 152)
There was a problem hiding this comment.
@lianetm I don't see any throttleMs in this file after the refactor. Could you please elaborate?
There was a problem hiding this comment.
you're right (I probably got mixed up with another file). All good.
| @@ -1609,15 +1608,19 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest { | |||
| // without describe permission on the topic, we shouldn't be able to fetch offsets | |||
| val offsetFetchRequest = createOffsetFetchRequestAllPartitions | |||
| var offsetFetchResponse = connectAndReceive[OffsetFetchResponse](offsetFetchRequest) | |||
| assertEquals(Errors.NONE, offsetFetchResponse.groupLevelError(group)) | |||
| assertTrue(offsetFetchResponse.partitionDataMap(group).isEmpty) | |||
There was a problem hiding this comment.
why removing this? this check is what ensures that we're not returning the offsets if no auth right?
There was a problem hiding this comment.
This is an oversight. Let me bring it back.
There was a problem hiding this comment.
we're still missing this assert it seems
There was a problem hiding this comment.
Pull Request Overview
This PR simplifies the OffsetFetchResponse by consolidating its constructors into a single data‐driven builder and updating its usage in both server and client code. Key changes include updating tests to use the new builder API, adjusting error handling logic based on response version, and refactoring legacy PartitionData usages.
Reviewed Changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala | Updated OffsetFetchResponse error checks and ACL validations to use the new builder API |
| core/src/main/scala/kafka/server/KafkaApis.scala | Replaced the direct constructor call with the new Builder approach for constructing OffsetFetchResponse |
| clients/... | Various test files and client modules have been updated to use the new OffsetFetchResponse.Builder and new error handling methods |
| clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java | Refactored internal error handling and the group() method logic to support multiple response versions |
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
Show resolved
Hide resolved
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
Show resolved
Hide resolved
While working on #19515, I came to
the conclusion that the OffsetFetchResponse is quite messy and overall
too complicated. This patch rationalize the constructors.
OffsetFetchResponse has a single constructor accepting the
OffsetFetchResponseData. A builder is introduced to handle the down
conversion. This will also simplify adding the topic ids. All the
changes are mechanical, replacing data structures by others.
Reviewers: Lianet Magrans lmagrans@confluent.io