[GitHub] [kafka] philipnee commented on a diff in pull request #12870: KAFKA-14367; Add `OffsetFetch` to the new `GroupCoordinator` interface

[GitBox <gi...@apache.org>]

philipnee commented on code in PR #12870:
URL: https://github.com/apache/kafka/pull/12870#discussion_r1054847935


##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -227,6 +227,24 @@ public boolean requireStable() {
         return data.requireStable();
     }
 
+    public List<OffsetFetchRequestData.OffsetFetchRequestGroup> groups() {
+        if (version() >= 8) {
+            return data.groups();
+        } else {

Review Comment:
   The else block is not needed.



##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1338,27 +1337,22 @@ class KafkaApis(val requestChannel: RequestChannel,
   /**
    * Handle an offset fetch request
    */
-  def handleOffsetFetchRequest(request: RequestChannel.Request): Unit = {
+  def handleOffsetFetchRequest(request: RequestChannel.Request): CompletableFuture[Unit] = {
     val version = request.header.apiVersion
     if (version == 0) {
-      // reading offsets from ZK
-      handleOffsetFetchRequestV0(request)
-    } else if (version >= 1 && version <= 7) {
-      // reading offsets from Kafka
-      handleOffsetFetchRequestBetweenV1AndV7(request)
+      handleOffsetFetchRequestFromZookeeper(request)

Review Comment:
   i think it can be more idiomatic and cleaner if we rewrite it using match - case 
   ```
   version match {
     case 0: handle...
     case  ...
   }
   ```



##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -227,6 +227,24 @@ public boolean requireStable() {
         return data.requireStable();
     }
 
+    public List<OffsetFetchRequestData.OffsetFetchRequestGroup> groups() {
+        if (version() >= 8) {
+            return data.groups();
+        } else {
+            OffsetFetchRequestData.OffsetFetchRequestGroup group =
+                new OffsetFetchRequestData.OffsetFetchRequestGroup().setGroupId(data.groupId());
+
+            data.topics().forEach(topic -> {

Review Comment:
   Do you think this can be a bit more intuitive?
   ```
   group.topics.addAll(
     data.topics().stream().map(new Offset...).collect(Collections.toList())
   )
   ```



##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java:
##########
@@ -208,6 +203,56 @@ public OffsetFetchResponse(int 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;
+
+            for (OffsetFetchResponseGroup group : data.groups()) {
+                this.groupLevelErrors.put(group.groupId(), Errors.forCode(group.errorCode()));
+            }

Review Comment:
   we could just return here to avoid the else. also save 1 indentation.



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org