dajac commented on code in PR #19642:
URL: https://github.com/apache/kafka/pull/19642#discussion_r2075128654
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java:
##########
@@ -60,221 +55,126 @@
public class OffsetFetchResponse extends AbstractResponse {
public static final long INVALID_OFFSET = -1L;
public static final String NO_METADATA = "";
- public static final PartitionData UNKNOWN_PARTITION = new
PartitionData(INVALID_OFFSET,
-
Optional.empty(),
-
NO_METADATA,
-
Errors.UNKNOWN_TOPIC_OR_PARTITION);
- public static final PartitionData UNAUTHORIZED_PARTITION = new
PartitionData(INVALID_OFFSET,
-
Optional.empty(),
-
NO_METADATA,
-
Errors.TOPIC_AUTHORIZATION_FAILED);
+
private static final List<Errors> PARTITION_ERRORS = Arrays.asList(
- Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.TOPIC_AUTHORIZATION_FAILED);
+ Errors.UNKNOWN_TOPIC_OR_PARTITION,
+ Errors.TOPIC_AUTHORIZATION_FAILED,
+ Errors.UNSTABLE_OFFSET_COMMIT
+ );
+ private final short version;
private final OffsetFetchResponseData data;
- private final Errors error;
- private final Map<String, Errors> groupLevelErrors = new HashMap<>();
-
- public static final class PartitionData {
- public final long offset;
- public final String metadata;
- public final Errors error;
- public final Optional<Integer> leaderEpoch;
+ // Lazily initialized when OffsetFetchResponse#group is called.
+ private Map<String, OffsetFetchResponseData.OffsetFetchResponseGroup>
groups = null;
- public PartitionData(long offset,
- Optional<Integer> leaderEpoch,
- String metadata,
- Errors error) {
- this.offset = offset;
- this.leaderEpoch = leaderEpoch;
- this.metadata = metadata;
- this.error = error;
- }
-
- public boolean hasError() {
- return this.error != Errors.NONE;
- }
-
- @Override
- public boolean equals(Object other) {
- if (!(other instanceof PartitionData))
- return false;
- PartitionData otherPartition = (PartitionData) other;
- return Objects.equals(this.offset, otherPartition.offset)
- && Objects.equals(this.leaderEpoch,
otherPartition.leaderEpoch)
- && Objects.equals(this.metadata, otherPartition.metadata)
- && Objects.equals(this.error, otherPartition.error);
- }
-
- @Override
- public String toString() {
- return "PartitionData("
- + "offset=" + offset
- + ", leaderEpoch=" +
leaderEpoch.orElse(NO_PARTITION_LEADER_EPOCH)
- + ", metadata=" + metadata
- + ", error='" + error.toString()
- + ")";
- }
+ public static class Builder {
+ private final List<OffsetFetchResponseGroup> groups;
- @Override
- public int hashCode() {
- return Objects.hash(offset, leaderEpoch, metadata, error);
+ public Builder(OffsetFetchResponseGroup group) {
+ this(List.of(group));
}
- }
-
- /**
- * Constructor without throttle time.
- * @param error Potential coordinator or group level error code (for api
version 2 and later)
- * @param responseData Fetched offset information grouped by
topic-partition
- */
- public OffsetFetchResponse(Errors error, Map<TopicPartition,
PartitionData> responseData) {
- this(DEFAULT_THROTTLE_TIME, error, responseData);
- }
- /**
- * Constructor with throttle time for version 0 to 7
- * @param throttleTimeMs The time in milliseconds that this response was
throttled
- * @param error Potential coordinator or group level error code (for api
version 2 and later)
- * @param responseData Fetched offset information grouped by
topic-partition
- */
- public OffsetFetchResponse(int throttleTimeMs, Errors error,
Map<TopicPartition, PartitionData> responseData) {
- super(ApiKeys.OFFSET_FETCH);
- Map<String, OffsetFetchResponseTopic> offsetFetchResponseTopicMap =
new HashMap<>();
- for (Map.Entry<TopicPartition, PartitionData> entry :
responseData.entrySet()) {
- String topicName = entry.getKey().topic();
- OffsetFetchResponseTopic topic =
offsetFetchResponseTopicMap.getOrDefault(
- topicName, new OffsetFetchResponseTopic().setName(topicName));
- PartitionData partitionData = entry.getValue();
- topic.partitions().add(new OffsetFetchResponsePartition()
-
.setPartitionIndex(entry.getKey().partition())
-
.setErrorCode(partitionData.error.code())
-
.setCommittedOffset(partitionData.offset)
- .setCommittedLeaderEpoch(
-
partitionData.leaderEpoch.orElse(NO_PARTITION_LEADER_EPOCH))
- .setMetadata(partitionData.metadata)
- );
- offsetFetchResponseTopicMap.put(topicName, topic);
+ public Builder(List<OffsetFetchResponseGroup> groups) {
+ this.groups = groups;
}
- this.data = new OffsetFetchResponseData()
- .setTopics(new ArrayList<>(offsetFetchResponseTopicMap.values()))
- .setErrorCode(error.code())
- .setThrottleTimeMs(throttleTimeMs);
- this.error = error;
- }
+ public OffsetFetchResponse build(short version) {
+ var data = new OffsetFetchResponseData();
- /**
- * Constructor with throttle time for version 8 and above.
- * @param throttleTimeMs The time in milliseconds that this response was
throttled
- * @param errors Potential coordinator or group level error code
- * @param responseData Fetched offset information grouped by
topic-partition and by group
- */
- public OffsetFetchResponse(int throttleTimeMs,
- Map<String, Errors> errors,
- Map<String, Map<TopicPartition, PartitionData>>
responseData) {
- super(ApiKeys.OFFSET_FETCH);
- List<OffsetFetchResponseGroup> groupList = new ArrayList<>();
- for (Entry<String, Map<TopicPartition, PartitionData>> entry :
responseData.entrySet()) {
- String groupName = entry.getKey();
- Map<TopicPartition, PartitionData> partitionDataMap =
entry.getValue();
- Map<String, OffsetFetchResponseTopics>
offsetFetchResponseTopicsMap = new HashMap<>();
- for (Entry<TopicPartition, PartitionData> partitionEntry :
partitionDataMap.entrySet()) {
- String topicName = partitionEntry.getKey().topic();
- OffsetFetchResponseTopics topic =
- offsetFetchResponseTopicsMap.getOrDefault(topicName,
- new OffsetFetchResponseTopics().setName(topicName));
- PartitionData partitionData = partitionEntry.getValue();
- topic.partitions().add(new OffsetFetchResponsePartitions()
- .setPartitionIndex(partitionEntry.getKey().partition())
- .setErrorCode(partitionData.error.code())
- .setCommittedOffset(partitionData.offset)
- .setCommittedLeaderEpoch(
-
partitionData.leaderEpoch.orElse(NO_PARTITION_LEADER_EPOCH))
- .setMetadata(partitionData.metadata));
- offsetFetchResponseTopicsMap.put(topicName, topic);
- }
- groupList.add(new OffsetFetchResponseGroup()
- .setGroupId(groupName)
- .setTopics(new
ArrayList<>(offsetFetchResponseTopicsMap.values()))
- .setErrorCode(errors.get(groupName).code()));
- groupLevelErrors.put(groupName, errors.get(groupName));
- }
- this.data = new OffsetFetchResponseData()
- .setGroups(groupList)
- .setThrottleTimeMs(throttleTimeMs);
- this.error = null;
- }
-
- public OffsetFetchResponse(List<OffsetFetchResponseGroup> groups, short
version) {
- super(ApiKeys.OFFSET_FETCH);
- data = new OffsetFetchResponseData();
-
- if (version >= 8) {
- data.setGroups(groups);
- error = null;
+ if (version >= BATCH_MIN_VERSION) {
+ data.setGroups(groups);
+ } else {
+ if (groups.size() != 1) {
+ throw new UnsupportedVersionException(
+ "Version " + version + " of OffsetFetchResponse only
supports one group."
+ );
+ }
- for (OffsetFetchResponseGroup group : data.groups()) {
- this.groupLevelErrors.put(group.groupId(),
Errors.forCode(group.errorCode()));
- }
- } else {
- if (groups.size() != 1) {
- throw new UnsupportedVersionException(
- "Version " + version + " of OffsetFetchResponse only
supports one group."
- );
+ OffsetFetchResponseGroup group = groups.get(0);
+ data.setErrorCode(group.errorCode());
+
+ group.topics().forEach(topic -> {
+ OffsetFetchResponseTopic newTopic = new
OffsetFetchResponseTopic().setName(topic.name());
+ data.topics().add(newTopic);
+
+ topic.partitions().forEach(partition -> {
+ OffsetFetchResponsePartition newPartition;
+
+ if (version <
TOP_LEVEL_ERROR_AND_NULL_TOPICS_MIN_VERSION && group.errorCode() !=
Errors.NONE.code()) {
+ // Versions prior to version 2 do not support a
top level error. Therefore,
+ // we put it at the partition level.
+ newPartition = new OffsetFetchResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(group.errorCode())
+ .setCommittedOffset(INVALID_OFFSET)
+ .setMetadata(NO_METADATA)
+
.setCommittedLeaderEpoch(NO_PARTITION_LEADER_EPOCH);
+ } else {
+ newPartition = new OffsetFetchResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(partition.errorCode())
+
.setCommittedOffset(partition.committedOffset())
+ .setMetadata(partition.metadata())
+
.setCommittedLeaderEpoch(partition.committedLeaderEpoch());
+ }
+
+ newTopic.partitions().add(newPartition);
+ });
+ });
}
- OffsetFetchResponseGroup group = groups.get(0);
- data.setErrorCode(group.errorCode());
- error = Errors.forCode(group.errorCode());
-
- group.topics().forEach(topic -> {
- OffsetFetchResponseTopic newTopic = new
OffsetFetchResponseTopic().setName(topic.name());
- data.topics().add(newTopic);
-
- topic.partitions().forEach(partition -> {
- OffsetFetchResponsePartition newPartition;
-
- if (version < 2 && group.errorCode() !=
Errors.NONE.code()) {
Review Comment:
https://issues.apache.org/jira/browse/KAFKA-19246
--
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
