chia7712 commented on code in PR #19572:
URL: https://github.com/apache/kafka/pull/19572#discussion_r2061327428
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -51,86 +49,9 @@ public static class Builder extends
AbstractRequest.Builder<OffsetFetchRequest>
public final OffsetFetchRequestData data;
Review Comment:
Given that you are refactoring this file, could you please change the
modifier from public to private? there is already a method `data()`
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -290,37 +198,49 @@ public OffsetFetchResponse getErrorResponse(Errors error)
{
}
Review Comment:
`getErrorResponse(Errors error)` is useless, so could you please remove it?
##########
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)
Review Comment:
Could you please replace `-1` by `RecordBatch.NO_PARTITION_LEADER_EPOCH` for
consistency?
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -34,15 +35,12 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Optional;
import java.util.stream.Collectors;
public class OffsetFetchRequest extends AbstractRequest {
private static final Logger log =
LoggerFactory.getLogger(OffsetFetchRequest.class);
- private static final List<OffsetFetchRequestTopic> ALL_TOPIC_PARTITIONS =
null;
private static final List<OffsetFetchRequestTopics>
ALL_TOPIC_PARTITIONS_BATCH = null;
Review Comment:
it seems this can be removed too, as it is used as null check .
--
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
