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

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

@@ -29,6 +29,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;
@@ -604,7 +605,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 13 drop topic name and add support to topic id. However, metadata can be used to map topic id to topic name.
+                    String topicName = metadata.topicNames().getOrDefault(r.topicId(), r.name());
+                    TopicPartition tp = new TopicPartition(topicName, p.index());
                     ProduceResponse.PartitionResponse partResp = new ProduceResponse.PartitionResponse(
                             Errors.forCode(p.errorCode()),
                             p.baseOffset(),
@@ -864,15 +867,21 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
             return;
 
         final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
+        Map<String, Uuid> topicIds = topicIdsForBatches(batches);
+
         ProduceRequestData.TopicProduceDataCollection tpd = new ProduceRequestData.TopicProduceDataCollection();
         for (ProducerBatch batch : batches) {
             TopicPartition tp = batch.topicPartition;
             MemoryRecords records = batch.records();
-            ProduceRequestData.TopicProduceData tpData = tpd.find(tp.topic());
+            Uuid topicId = topicIds.get(tp.topic());
+            ProduceRequestData.TopicProduceData tpData = tpd.find(tp.topic(), topicId);
+
             if (tpData == null) {
-                tpData = new ProduceRequestData.TopicProduceData().setName(tp.topic());
+                tpData = new ProduceRequestData.TopicProduceData()
+                        .setTopicId(topicId).setName(tp.topic());
                 tpd.add(tpData);
             }
+
             tpData.partitionData().add(new ProduceRequestData.PartitionProduceData()
                     .setIndex(tp.partition())
                     .setRecords(records));
@@ -903,6 +912,15 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
         log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
     }
 
+    private Map<String, Uuid> topicIdsForBatches(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,7 @@
 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.errors.UnsupportedCompressionTypeException;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.ProduceResponseData;
@@ -96,7 +96,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);
@@ -107,15 +107,20 @@ 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()),
+                            // While topic id and name might not be populated at the same time in the request all the time;
+                            // for example on server side they will never be populated together while in produce client they will be,
+                            // to simplify initializing `TopicIdPartition` the code will use both topic name and id.
+                            // TopicId will be Uuid.ZERO_UUID in versions < 13 and topic name will be used as main identifier of topic partition.
+                            // TopicName will be empty string in versions >= 13 and topic id will be used as the main identifier.
+                            tmpPartitionSizes.compute(new TopicIdPartition(topicData.topicId(), partitionData.index(), topicData.name()),
                                 (ignored, previousValue) ->
                                     partitionData.records().sizeInBytes() + (previousValue == null ? 0 : previousValue))
                         )
@@ -161,14 +166,14 @@ public ProduceResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         if (acks == 0) return null;
         ApiError apiError = ApiError.fromThrowable(e);
         ProduceResponseData data = new ProduceResponseData().setThrottleTimeMs(throttleTimeMs);
-        partitionSizes().forEach((tp, ignored) -> {
-            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tp.topic());
+        partitionSizes().forEach((tpId, ignored) -> {
+            ProduceResponseData.TopicProduceResponse tpr = data.responses().find(tpId.topic(), tpId.topicId());
             if (tpr == null) {
-                tpr = new ProduceResponseData.TopicProduceResponse().setName(tp.topic());
+                tpr = new ProduceResponseData.TopicProduceResponse().setName(tpId.topic()).setTopicId(tpId.topicId());
                 data.responses().add(tpr);
             }
             tpr.partitionResponses().add(new ProduceResponseData.PartitionProduceResponse()
-                    .setIndex(tp.partition())
+                    .setIndex(tpId.partition())
                     .setRecordErrors(Collections.emptyList())
                     .setBaseOffset(INVALID_OFFSET)
                     .setLogAppendTimeMs(RecordBatch.NO_TIMESTAMP)

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;
@@ -56,6 +56,7 @@
  * {@link Errors#INVALID_TXN_STATE}
  * {@link Errors#INVALID_PRODUCER_ID_MAPPING}
  * {@link Errors#CONCURRENT_TRANSACTIONS}
+ * {@link Errors#UNKNOWN_TOPIC_ID}
  */
 public class ProduceResponse extends AbstractResponse {
     public static final long INVALID_OFFSET = -1L;
@@ -73,7 +74,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());
     }
 
@@ -84,7 +85,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()));
     }
 
@@ -97,16 +98,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

@@ -1690,6 +1690,7 @@ public static ConfigDef mergeConfigs(List<ConfigDef> configDefs) {
         configDefs.forEach(configDef -> configDef.configKeys().values().forEach(all::define));
         return all;
     }
+
     /**
      * A runnable that can throw checked exception.
      */

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

@@ -46,7 +46,8 @@
   // transaction V2 (KIP_890 part 2) is enabled, the produce request will also include the function for a
   // AddPartitionsToTxn call. If V2 is disabled, the client can't use produce request version higher than 11 within
   // a transaction.
-  "validVersions": "3-12",
+  // Version 13 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "3-13",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "TransactionalId", "type": "string", "versions": "3+", "nullableVersions": "3+", "default": "null", "entityType": "transactionalId",
@@ -57,8 +58,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-12", "entityType": "topicName", "mapKey": true, "ignorable": true,
         "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "13+", "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

@@ -40,13 +40,15 @@
   // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
   //
   // Version 12 is the same as version 10 (KIP-890).
-  "validVersions": "3-12",
+  // Version 13 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "3-13",
   "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-12", "entityType": "topicName", "mapKey": true, "ignorable": true,
         "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "13+", "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+",

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -38,6 +38,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.config.ConfigException;
@@ -1424,7 +1425,7 @@ public void testTransactionV2ProduceWithConcurrentTransactionError() throws Exce
         KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
 
         String topic = "foo";
-        TopicPartition topicPartition = new TopicPartition(topic, 0);
+        TopicIdPartition topicIdPartition = new TopicIdPartition(Uuid.randomUuid(), 0, topic);
         Cluster cluster = TestUtils.singletonCluster(topic, 1);
 
         when(ctx.sender.isRunning()).thenReturn(true);
@@ -1463,8 +1464,8 @@ public void testTransactionV2ProduceWithConcurrentTransactionError() throws Exce
 
         client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some-txn", NODE));
         client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
-        client.prepareResponse(produceResponse(topicPartition, 1L, Errors.CONCURRENT_TRANSACTIONS, 0, 1));
-        client.prepareResponse(produceResponse(topicPartition, 1L, Errors.NONE, 0, 1));
+        client.prepareResponse(produceResponse(topicIdPartition, 1L, Errors.CONCURRENT_TRANSACTIONS, 0, 1));
+        client.prepareResponse(produceResponse(topicIdPartition, 1L, Errors.NONE, 0, 1));
         client.prepareResponse(endTxnResponse(Errors.NONE));
 
         try (KafkaProducer<String, String> producer = new KafkaProducer<>(
@@ -1750,6 +1751,7 @@ public void testTransactionV2Produce() throws Exception {
         KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
 
         String topic = "foo";
+        Uuid topicId = Uuid.fromString("klZ9sa2rSvig6QpgGXzALT");
         TopicPartition topicPartition = new TopicPartition(topic, 0);
         Cluster cluster = TestUtils.singletonCluster(topic, 1);
 
@@ -1789,7 +1791,7 @@ public void testTransactionV2Produce() throws Exception {
 
         client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some-txn", NODE));
         client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
-        client.prepareResponse(produceResponse(topicPartition, 1L, Errors.NONE, 0, 1));
+        client.prepareResponse(produceResponse(new TopicIdPartition(topicId, topicPartition), 1L, Errors.NONE, 0, 1));
         client.prepareResponse(endTxnResponse(Errors.NONE));
 
         try (KafkaProducer<String, String> producer = new KafkaProducer<>(
@@ -2649,9 +2651,9 @@ void testDeliveryTimeoutAndLingerMsConfig() {
     }
 
     @SuppressWarnings("deprecation")
-    private ProduceResponse produceResponse(TopicPartition tp, long offset, Errors error, int throttleTimeMs, int logStartOffset) {
+    private ProduceResponse produceResponse(TopicIdPartition topicIdPartition, long offset, Errors error, int throttleTimeMs, int logStartOffset) {
         ProduceResponse.PartitionResponse resp = new ProduceResponse.PartitionResponse(error, offset, RecordBatch.NO_TIMESTAMP, logStartOffset);
-        Map<TopicPartition, ProduceResponse.PartitionResponse> partResp = singletonMap(tp, resp);
+        Map<TopicIdPartition, ProduceResponse.PartitionResponse> partResp = singletonMap(topicIdPartition, resp);
         return new ProduceResponse(partResp, throttleTimeMs);
     }