frankvicky commented on code in PR #19572:
URL: https://github.com/apache/kafka/pull/19572#discussion_r2061311295
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -290,37 +198,49 @@ public OffsetFetchResponse getErrorResponse(Errors error)
{
}
public OffsetFetchResponse getErrorResponse(int throttleTimeMs, Errors
error) {
- Map<TopicPartition, OffsetFetchResponse.PartitionData>
responsePartitions = new HashMap<>();
if (version() < 2) {
- OffsetFetchResponse.PartitionData partitionError = new
OffsetFetchResponse.PartitionData(
- OffsetFetchResponse.INVALID_OFFSET,
- Optional.empty(),
- OffsetFetchResponse.NO_METADATA,
- error);
-
- for (OffsetFetchRequestTopic topic : this.data.topics()) {
- for (int partitionIndex : topic.partitionIndexes()) {
- responsePartitions.put(
- new TopicPartition(topic.name(), partitionIndex),
partitionError);
- }
- }
- return new OffsetFetchResponse(error, responsePartitions);
- }
- if (version() == 2) {
- return new OffsetFetchResponse(error, responsePartitions);
- }
- if (version() >= 3 && version() < 8) {
- return new OffsetFetchResponse(throttleTimeMs, error,
responsePartitions);
- }
- List<String> groupIds = groupIds();
- Map<String, Errors> errorsMap = new HashMap<>(groupIds.size());
- Map<String, Map<TopicPartition, OffsetFetchResponse.PartitionData>>
partitionMap =
- new HashMap<>(groupIds.size());
- for (String g : groupIds) {
- errorsMap.put(g, error);
- partitionMap.put(g, responsePartitions);
+ // The response does not support top level error so we return each
+ // partition with the error.
+ return new OffsetFetchResponse(
+ new OffsetFetchResponseData()
+ .setThrottleTimeMs(throttleTimeMs)
+ .setTopics(data.topics().stream().map(topic ->
+ new OffsetFetchResponseData.OffsetFetchResponseTopic()
+ .setName(topic.name())
+
.setPartitions(topic.partitionIndexes().stream().map(partition ->
+ new
OffsetFetchResponseData.OffsetFetchResponsePartition()
+ .setPartitionIndex(partition)
+ .setErrorCode(error.code())
+
.setCommittedOffset(OffsetFetchResponse.INVALID_OFFSET)
+
.setMetadata(OffsetFetchResponse.NO_METADATA)
+ .setCommittedLeaderEpoch(-1)
+ ).collect(Collectors.toList()))
+ ).collect(Collectors.toList())),
+ version()
+ );
+ } else if (version() < 8) {
Review Comment:
Do we need some statics variables for these magic numbers?
--
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
