Hangleton commented on code in PR #13240:
URL: https://github.com/apache/kafka/pull/13240#discussion_r1121724366
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java:
##########
@@ -50,14 +52,21 @@
public class OffsetCommitResponse extends AbstractResponse {
private final OffsetCommitResponseData data;
+ private final short version;
Review Comment:
Adding the version to the response seems to be an anti-pattern as I haven't
seen any other similar use in other responses. Semantically it should be OK
because the response instance is supposed to be built against a given version.
If another approach is advisable, I will remove it.
##########
clients/src/test/java/org/apache/kafka/common/message/MessageTest.java:
##########
@@ -435,17 +435,32 @@ public void testOffsetCommitRequestVersions() throws
Exception {
int partition = 2;
int offset = 100;
- testAllMessageRoundTrips(new OffsetCommitRequestData()
- .setGroupId(groupId)
- .setTopics(Collections.singletonList(
- new OffsetCommitRequestTopic()
- .setName(topicName)
-
.setPartitions(Collections.singletonList(
- new
OffsetCommitRequestPartition()
-
.setPartitionIndex(partition)
-
.setCommittedMetadata(metadata)
-
.setCommittedOffset(offset)
- )))));
+ OffsetCommitRequestData byTopicName = new OffsetCommitRequestData()
+ .setGroupId(groupId)
+ .setTopics(Collections.singletonList(
+ new OffsetCommitRequestTopic()
+ .setName(topicName)
+ .setPartitions(Collections.singletonList(
+ new OffsetCommitRequestPartition()
+ .setPartitionIndex(partition)
+ .setCommittedMetadata(metadata)
+ .setCommittedOffset(offset)
+ ))));
+
+ OffsetCommitRequestData byTopicId = new OffsetCommitRequestData()
+ .setGroupId(groupId)
+ .setTopics(Collections.singletonList(
+ new OffsetCommitRequestTopic()
+ .setTopicId(Uuid.randomUuid())
+ .setPartitions(Collections.singletonList(
+ new OffsetCommitRequestPartition()
+ .setPartitionIndex(partition)
+ .setCommittedMetadata(metadata)
+ .setCommittedOffset(offset)
+ ))));
+
+ testAllMessageRoundTripsBeforeVersion((short) 9, byTopicName,
byTopicName);
Review Comment:
Note: is this OK to break message round trip between < 9 and >= 9?
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java:
##########
@@ -188,12 +199,62 @@ public Builder merge(
}
});
}
-
return this;
}
- public OffsetCommitResponse build() {
- return new OffsetCommitResponse(data);
+ public final OffsetCommitResponse build() {
+ return new OffsetCommitResponse(data, version);
+ }
+
+ protected abstract void validate(String topicName, Uuid topicId);
+
+ protected abstract T classifer(String topicName, Uuid topicId);
+
+ private OffsetCommitResponseTopic getOrCreateTopic(String topicName,
Uuid topicId) {
+ T topicKey = classifer(topicName, topicId);
+ OffsetCommitResponseTopic topic = topics.get(topicKey);
+ if (topic == null) {
+ topic = new
OffsetCommitResponseTopic().setName(topicName).setTopicId(topicId);
+ data.topics().add(topic);
+ topics.put(topicKey, topic);
+ }
+ return topic;
+ }
+ }
+
+ public static final class BuilderByTopicId extends Builder<Uuid> {
Review Comment:
Will add Javadoc.
--
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
