KAFKA-10551: Add topic id support to produce request and response

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -33,6 +33,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.FencedLeaderEpochException;
@@ -610,7 +611,9 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
                 // This will be set by completeBatch.
                 Map<TopicPartition, Metadata.LeaderIdAndEpoch> partitionsWithUpdatedLeaderInfo = new HashMap<>();
                 produceResponse.data().responses().forEach(r -> r.partitionResponses().forEach(p -> {
-                    TopicPartition tp = new TopicPartition(r.name(), p.index());
+                    // Version 12 drop topic name and add support to topic id. However, metadata can be used to map topic id to topic name.
+                    String topicName = (r.name() == null || r.name().isEmpty()) ? metadata.topicNames().get(r.topicId()) : r.name();
+                    TopicPartition tp = new TopicPartition(topicName, p.index());
                     ProduceResponse.PartitionResponse partResp = new ProduceResponse.PartitionResponse(
                             Errors.forCode(p.errorCode()),
                             p.baseOffset(),
@@ -877,6 +880,9 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
             if (batch.magic() < minUsedMagic)
                 minUsedMagic = batch.magic();
         }
+        Map<String, Uuid> topicIds = getTopicIdsFromBatches(batches);
+        boolean canUseTopicId = !topicIds.entrySet().stream().anyMatch(e -> e.getValue() == Uuid.ZERO_UUID);
+
         ProduceRequestData.TopicProduceDataCollection tpd = new ProduceRequestData.TopicProduceDataCollection();
         for (ProducerBatch batch : batches) {
             TopicPartition tp = batch.topicPartition;
@@ -891,11 +897,15 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
             // which is supporting the new magic version to one which doesn't, then we will need to convert.
             if (!records.hasMatchingMagic(minUsedMagic))
                 records = batch.records().downConvert(minUsedMagic, 0, time).records();
-            ProduceRequestData.TopicProduceData tpData = tpd.find(tp.topic());
-            if (tpData == null) {
-                tpData = new ProduceRequestData.TopicProduceData().setName(tp.topic());
-                tpd.add(tpData);
+            Optional<ProduceRequestData.TopicProduceData> topicProduceData = canUseTopicId ?
+                    Optional.ofNullable(tpd.find(tp.topic(), topicIds.get(tp.topic()))) :
+                    tpd.stream().filter(data -> data.name().equals(tp.topic())).findFirst();
+
+            ProduceRequestData.TopicProduceData tpData = topicProduceData.orElse(new ProduceRequestData.TopicProduceData().setName(tp.topic()));
+            if (canUseTopicId) {
+                tpData.setTopicId(topicIds.get(tp.topic()));
             }
+            tpd.add(tpData);
             tpData.partitionData().add(new ProduceRequestData.PartitionProduceData()
                     .setIndex(tp.partition())
                     .setRecords(records));
@@ -922,6 +932,15 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
         log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
     }
 
+    private Map<String, Uuid> getTopicIdsFromBatches(List<ProducerBatch> batches) {
+        return batches.stream()
+                .collect(Collectors.toMap(
+                        b -> b.topicPartition.topic(),
+                        b -> metadata.topicIds().getOrDefault(b.topicPartition.topic(), Uuid.ZERO_UUID),
+                        (existing, replacement) -> replacement)
+                );
+    }
+
     /**
      * Wake up the selector associated with this send thread
      */

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

@@ -17,7 +17,8 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.InvalidRecordException;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedCompressionTypeException;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
@@ -51,13 +52,20 @@ public static Builder forMagic(byte magic, ProduceRequestData data) {
         if (magic < RecordBatch.MAGIC_VALUE_V2) {
             minVersion = 2;
             maxVersion = 2;
+        } else if (canNotSupportTopicId(data)) {
+            minVersion = 3;
+            maxVersion = 11;
         } else {
             minVersion = 3;
             maxVersion = ApiKeys.PRODUCE.latestVersion();
         }
         return new Builder(minVersion, maxVersion, data);
     }
 
+    private static boolean canNotSupportTopicId(ProduceRequestData data) {
+        return data.topicData().stream().anyMatch(d -> d.topicId() == null || d.topicId() == Uuid.ZERO_UUID);
+    }
+
     public static Builder forCurrentMagic(ProduceRequestData data) {
         return forMagic(RecordBatch.CURRENT_MAGIC_VALUE, data);
     }
@@ -117,7 +125,7 @@ public String toString() {
     // Care should be taken in methods that use this field.
     private volatile ProduceRequestData data;
     // the partitionSizes is lazily initialized since it is used by server-side in production.
-    private volatile Map<TopicPartition, Integer> partitionSizes;
+    private volatile Map<TopicIdPartition, Integer> partitionSizes;
 
     public ProduceRequest(ProduceRequestData produceRequestData, short version) {
         super(ApiKeys.PRODUCE, version);
@@ -128,15 +136,15 @@ public ProduceRequest(ProduceRequestData produceRequestData, short version) {
     }
 
     // visible for testing
-    Map<TopicPartition, Integer> partitionSizes() {
+    Map<TopicIdPartition, Integer> partitionSizes() {
         if (partitionSizes == null) {
             // this method may be called by different thread (see the comment on data)
             synchronized (this) {
                 if (partitionSizes == null) {
-                    Map<TopicPartition, Integer> tmpPartitionSizes = new HashMap<>();
+                    Map<TopicIdPartition, Integer> tmpPartitionSizes = new HashMap<>();
                     data.topicData().forEach(topicData ->
                         topicData.partitionData().forEach(partitionData ->
-                            tmpPartitionSizes.compute(new TopicPartition(topicData.name(), partitionData.index()),
+                            tmpPartitionSizes.compute(new TopicIdPartition(topicData.topicId(), partitionData.index(), topicData.name()),
                                 (ignored, previousValue) ->
                                     partitionData.records().sizeInBytes() + (previousValue == null ? 0 : previousValue))
                         )
@@ -183,9 +191,9 @@ public ProduceResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         ApiError apiError = ApiError.fromThrowable(e);
         ProduceResponseData data = new ProduceResponseData().setThrottleTimeMs(throttleTimeMs);
         partitionSizes().forEach((tp, ignored) -> {
-            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic());
+            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic(), tp.topicId());
             if (tpr == null) {
-                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic());
+                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic()).setTopicId(tp.topicId());
                 data.responses().add(tpr);
             }
             tpr.partitionResponses().add(new ProduceResponseData.PartitionProduceResponse()

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

@@ -17,7 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.Node;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.ProduceResponseData.LeaderIdAndEpoch;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -33,6 +33,7 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+
 /**
  * This wrapper supports both v0 and v8 of ProduceResponse.
  *
@@ -72,7 +73,7 @@ public ProduceResponse(ProduceResponseData produceResponseData) {
      * @param responses Produced data grouped by topic-partition
      */
     @Deprecated
-    public ProduceResponse(Map<TopicPartition, PartitionResponse> responses) {
+    public ProduceResponse(Map<TopicIdPartition, PartitionResponse> responses) {
         this(responses, DEFAULT_THROTTLE_TIME, Collections.emptyList());
     }
 
@@ -83,7 +84,7 @@ public ProduceResponse(Map<TopicPartition, PartitionResponse> responses) {
      * @param throttleTimeMs Time in milliseconds the response was throttled
      */
     @Deprecated
-    public ProduceResponse(Map<TopicPartition, PartitionResponse> responses, int throttleTimeMs) {
+    public ProduceResponse(Map<TopicIdPartition, PartitionResponse> responses, int throttleTimeMs) {
         this(toData(responses, throttleTimeMs, Collections.emptyList()));
     }
 
@@ -96,16 +97,16 @@ public ProduceResponse(Map<TopicPartition, PartitionResponse> responses, int thr
      * @param nodeEndpoints List of node endpoints
      */
     @Deprecated
-    public ProduceResponse(Map<TopicPartition, PartitionResponse> responses, int throttleTimeMs, List<Node> nodeEndpoints) {
+    public ProduceResponse(Map<TopicIdPartition, PartitionResponse> responses, int throttleTimeMs, List<Node> nodeEndpoints) {
         this(toData(responses, throttleTimeMs, nodeEndpoints));
     }
 
-    private static ProduceResponseData toData(Map<TopicPartition, PartitionResponse> responses, int throttleTimeMs, List<Node> nodeEndpoints) {
+    private static ProduceResponseData toData(Map<TopicIdPartition, PartitionResponse> responses, int throttleTimeMs, List<Node> nodeEndpoints) {
         ProduceResponseData data = new ProduceResponseData().setThrottleTimeMs(throttleTimeMs);
         responses.forEach((tp, response) -> {
-            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic());
+            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic(), tp.topicId());
             if (tpr == null) {
-                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic());
+                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic()).setTopicId(tp.topicId());
                 data.responses().add(tpr);
             }
             tpr.partitionResponses()

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -1654,6 +1654,15 @@ public static void require(boolean requirement) {
             throw new IllegalArgumentException("requirement failed");
     }
 
+    /**
+     * Convert a map's keys to another type.
+     */
+    public static <K1, K2, V> Map<K2, V> convertKeys(Map<K1, V> originalMap, Function<K1, K2> converter) {
+        Map<K2, V> newMap = new HashMap<>();
+        originalMap.forEach((key, value) -> newMap.put(converter.apply(key), value));
+        return newMap;
+    }
+
     /**
      * A runnable that can throw checked exception.
      */

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

@@ -37,7 +37,8 @@
   // Version 10 is the same as version 9 (KIP-951).
   //
   // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
-  "validVersions": "0-11",
+  // Version 12 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "0-12",
   "deprecatedVersions": "0-6",
   "flexibleVersions": "9+",
   "fields": [
@@ -49,8 +50,9 @@
       "about": "The timeout to await a response in milliseconds." },
     { "name": "TopicData", "type": "[]TopicProduceData", "versions": "0+",
       "about": "Each topic to produce to.", "fields": [
-      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", "mapKey": true,
+      { "name": "Name", "type": "string", "versions": "0-11", "entityType": "topicName", "mapKey": true, "ignorable": true,
         "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "12+", "mapKey": true, "ignorable": true, "about": "The unique topic ID" },
       { "name": "PartitionData", "type": "[]PartitionProduceData", "versions": "0+",
         "about": "Each partition to produce to.", "fields": [
         { "name": "Index", "type": "int32", "versions": "0+",

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

@@ -36,13 +36,15 @@
   // Version 10 adds 'CurrentLeader' and 'NodeEndpoints' as tagged fields (KIP-951)
   //
   // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
-  "validVersions": "0-11",
+  // Version 12 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "0-12",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "Responses", "type": "[]TopicProduceResponse", "versions": "0+",
       "about": "Each produce response", "fields": [
-      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", "mapKey": true,
+      { "name": "Name", "type": "string", "versions": "0-11", "entityType": "topicName", "mapKey": true, "ignorable": true,
         "about": "The topic name" },
+      { "name": "TopicId", "type": "uuid", "versions": "12+", "mapKey": true, "ignorable": true, "about": "The unique topic ID" },
       { "name": "PartitionResponses", "type": "[]PartitionProduceResponse", "versions": "0+",
         "about": "Each partition that we produced to within the topic.", "fields": [
         { "name": "Index", "type": "int32", "versions": "0+",