wip

Author: dajac

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.OffsetFetchRequestData;
 import org.apache.kafka.common.message.OffsetFetchRequestData.OffsetFetchRequestGroup;
@@ -71,7 +72,8 @@ public Builder(String groupId,
                        boolean requireStable,
                        List<TopicPartition> partitions,
                        boolean throwOnFetchStableOffsetsUnsupported) {
-            super(ApiKeys.OFFSET_FETCH);
+            // It can only be used with topic names.
+            super(ApiKeys.OFFSET_FETCH, ApiKeys.OFFSET_FETCH.oldestVersion(), (short) 9);
 
             OffsetFetchRequestData.OffsetFetchRequestGroup group =
                 new OffsetFetchRequestData.OffsetFetchRequestGroup()
@@ -103,7 +105,8 @@ public Builder(String groupId,
         public Builder(Map<String, List<TopicPartition>> groupIdToTopicPartitionMap,
                        boolean requireStable,
                        boolean throwOnFetchStableOffsetsUnsupported) {
-            super(ApiKeys.OFFSET_FETCH);
+            // It can only be used with topic names.
+            super(ApiKeys.OFFSET_FETCH, ApiKeys.OFFSET_FETCH.oldestVersion(), (short) 9);
 
             List<OffsetFetchRequestGroup> groups = new ArrayList<>();
             for (Entry<String, List<TopicPartition>> entry : groupIdToTopicPartitionMap.entrySet()) {
@@ -134,6 +137,12 @@ public Builder(Map<String, List<TopicPartition>> groupIdToTopicPartitionMap,
             this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported;
         }
 
+        public Builder(OffsetFetchRequestData data, boolean throwOnFetchStableOffsetsUnsupported) {
+            super(ApiKeys.OFFSET_FETCH);
+            this.data = data;
+            this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported;
+        }
+
         @Override
         public OffsetFetchRequest build(short version) {
             if (data.groups().size() > 1 && version < 8) {

clients/src/main/resources/common/message/OffsetFetchRequest.json

@@ -63,7 +63,7 @@
         "about": "The member epoch if using the new consumer protocol (KIP-848)." },
       { "name": "Topics", "type": "[]OffsetFetchRequestTopics", "versions": "8+", "nullableVersions": "8+",
         "about": "Each topic we would like to fetch offsets for, or null to fetch offsets for all topics.", "fields": [
-        { "name": "Name", "type": "string", "versions": "8+", "entityType": "topicName", "ignorable": true,
+        { "name": "Name", "type": "string", "versions": "8-9", "entityType": "topicName", "ignorable": true,
           "about": "The topic name."},
         { "name":  "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
           "about": "The topic ID." },

clients/src/main/resources/common/message/OffsetFetchResponse.json

@@ -81,7 +81,7 @@
         "about": "The group ID." },
       { "name": "Topics", "type": "[]OffsetFetchResponseTopics", "versions": "8+",
         "about": "The responses per topic.", "fields": [
-        { "name": "Name", "type": "string", "versions": "8+", "entityType": "topicName", "ignorable": true,
+        { "name": "Name", "type": "string", "versions": "8-9", "entityType": "topicName", "ignorable": true,
           "about": "The topic name." },
         { "name":  "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
           "about": "The topic ID." },

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -1056,7 +1056,7 @@ class KafkaApis(val requestChannel: RequestChannel,
               // If the topic is not provided by the group coordinator, we set it
               // using the metadata cache.
               if (topic.topicId == Uuid.ZERO_UUID) {
-                metadataCache.getTopicName(topic.topicId).ifPresent(name => topic.setName(name))
+                topic.setTopicId(metadataCache.getTopicId(topic.name))
               }
               // If we don't have the topic id at all, we skip the topic because
               // we can not serialize it without it.