skaundinya15 commented on a change in pull request #10962: URL: https://github.com/apache/kafka/pull/10962#discussion_r664798862
########## File path: clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java ########## @@ -78,26 +85,107 @@ boolean isAllTopicPartitions() { return this.data.topics() == ALL_TOPIC_PARTITIONS; } + public Builder(Map<String, List<TopicPartition>> groupIdToTopicPartitionMap, + boolean requireStable, + boolean throwOnFetchStableOffsetsUnsupported) { + super(ApiKeys.OFFSET_FETCH); + + List<OffsetFetchRequestGroup> groups = new ArrayList<>(); + for (Entry<String, List<TopicPartition>> entry : groupIdToTopicPartitionMap.entrySet()) { + String groupName = entry.getKey(); + List<TopicPartition> tpList = entry.getValue(); + final List<OffsetFetchRequestTopics> topics; + if (tpList != null) { + Map<String, OffsetFetchRequestTopics> offsetFetchRequestTopicMap = + new HashMap<>(); + for (TopicPartition topicPartition : tpList) { + String topicName = topicPartition.topic(); + OffsetFetchRequestTopics topic = offsetFetchRequestTopicMap.getOrDefault( + topicName, new OffsetFetchRequestTopics().setName(topicName)); + topic.partitionIndexes().add(topicPartition.partition()); + offsetFetchRequestTopicMap.put(topicName, topic); + } + topics = new ArrayList<>(offsetFetchRequestTopicMap.values()); + } else { + topics = ALL_TOPIC_PARTITIONS_BATCH; + } + groups.add(new OffsetFetchRequestGroup() + .setGroupId(groupName) + .setTopics(topics)); + } + this.data = new OffsetFetchRequestData() + .setGroupIds(groups) + .setRequireStable(requireStable); + this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported; + } + @Override public OffsetFetchRequest build(short version) { if (isAllTopicPartitions() && version < 2) { throw new UnsupportedVersionException("The broker only supports OffsetFetchRequest " + "v" + version + ", but we need v2 or newer to request all topic partitions."); } - + if (data.groupIds().size() > 1 && version < 8) { + throw new NoBatchedOffsetFetchRequestException("Broker does not support" + + " batching groups for fetch offset request on version " + version); + } if (data.requireStable() && version < 7) { if (throwOnFetchStableOffsetsUnsupported) { throw new UnsupportedVersionException("Broker unexpectedly " + "doesn't support requireStable flag on version " + version); } else { log.trace("Fallback the requireStable flag to false as broker " + - "only supports OffsetFetchRequest version {}. Need " + - "v7 or newer to enable this feature", version); + "only supports OffsetFetchRequest version {}. Need " + + "v7 or newer to enable this feature", version); return new OffsetFetchRequest(data.setRequireStable(false), version); } } - + if (version < 8) { + OffsetFetchRequestData oldDataFormat = null; + if (!data.groupIds().isEmpty()) { + OffsetFetchRequestGroup group = data.groupIds().get(0); + String groupName = group.groupId(); + List<OffsetFetchRequestTopics> topics = group.topics(); + List<OffsetFetchRequestTopic> oldFormatTopics = null; + if (topics != null) { + oldFormatTopics = topics + .stream() + .map(t -> + new OffsetFetchRequestTopic() + .setName(t.name()) + .setPartitionIndexes(t.partitionIndexes())) + .collect(Collectors.toList()); + } + oldDataFormat = new OffsetFetchRequestData() + .setGroupId(groupName) + .setTopics(oldFormatTopics) + .setRequireStable(data.requireStable()); + } + return new OffsetFetchRequest(oldDataFormat == null ? data : oldDataFormat, version); + } + // version 8 but have used old format of request, convert to version 8 of request Review comment: Ah okay, yes that makes sense, will make the change. -- 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