KAFKA-10547; add topicId in MetadataResp

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -1788,7 +1788,7 @@ void handleResponse(AbstractResponse abstractResponse) {
                     }
                     partitions.sort(Comparator.comparingInt(TopicPartitionInfo::partition));
                     TopicDescription topicDescription = new TopicDescription(topicName, isInternal, partitions,
-                        validAclOperations(response.topicAuthorizedOperations(topicName).get()));
+                        validAclOperations(response.topicAuthorizedOperations(topicName).get()), cluster.topicId(topicName));
                     future.complete(topicDescription);
                 }
             }

clients/src/main/java/org/apache/kafka/clients/admin/TopicDescription.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.clients.admin;
 
 import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.utils.Utils;
 
@@ -34,6 +35,7 @@ public class TopicDescription {
     private final boolean internal;
     private final List<TopicPartitionInfo> partitions;
     private final Set<AclOperation> authorizedOperations;
+    private final Uuid topicId;
 
     @Override
     public boolean equals(final Object o) {
@@ -74,10 +76,16 @@ public TopicDescription(String name, boolean internal, List<TopicPartitionInfo>
      */
     public TopicDescription(String name, boolean internal, List<TopicPartitionInfo> partitions,
                             Set<AclOperation> authorizedOperations) {
+        this(name, internal, partitions, authorizedOperations, Uuid.ZERO_UUID);
+    }
+
+    public TopicDescription(String name, boolean internal, List<TopicPartitionInfo> partitions,
+                            Set<AclOperation> authorizedOperations, Uuid topicId) {
         this.name = name;
         this.internal = internal;
         this.partitions = partitions;
         this.authorizedOperations = authorizedOperations;
+        this.topicId = topicId;
     }
 
     /**
@@ -95,6 +103,10 @@ public boolean isInternal() {
         return internal;
     }
 
+    public Uuid topicId() {
+        return topicId;
+    }
+
     /**
      * A list of partitions where the index represents the partition id and the element contains leadership and replica
      * information for that partition.

clients/src/main/java/org/apache/kafka/common/Cluster.java

@@ -46,6 +46,7 @@ public final class Cluster {
     private final Map<Integer, List<PartitionInfo>> partitionsByNode;
     private final Map<Integer, Node> nodesById;
     private final ClusterResource clusterResource;
+    private final Map<String, Uuid> topicIds;
 
     /**
      * Create a new cluster with the given id, nodes and partitions
@@ -57,7 +58,7 @@ public Cluster(String clusterId,
                    Collection<PartitionInfo> partitions,
                    Set<String> unauthorizedTopics,
                    Set<String> internalTopics) {
-        this(clusterId, false, nodes, partitions, unauthorizedTopics, Collections.emptySet(), internalTopics, null);
+        this(clusterId, false, nodes, partitions, unauthorizedTopics, Collections.emptySet(), internalTopics, null, Collections.emptyMap());
     }
 
     /**
@@ -71,7 +72,7 @@ public Cluster(String clusterId,
                    Set<String> unauthorizedTopics,
                    Set<String> internalTopics,
                    Node controller) {
-        this(clusterId, false, nodes, partitions, unauthorizedTopics, Collections.emptySet(), internalTopics, controller);
+        this(clusterId, false, nodes, partitions, unauthorizedTopics, Collections.emptySet(), internalTopics, controller, Collections.emptyMap());
     }
 
     /**
@@ -86,7 +87,23 @@ public Cluster(String clusterId,
                    Set<String> invalidTopics,
                    Set<String> internalTopics,
                    Node controller) {
-        this(clusterId, false, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller);
+        this(clusterId, false, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller, Collections.emptyMap());
+    }
+
+    /**
+     * Create a new cluster with the given id, nodes, partitions and topicIds
+     * @param nodes The nodes in the cluster
+     * @param partitions Information about a subset of the topic-partitions this cluster hosts
+     */
+    public Cluster(String clusterId,
+                   Collection<Node> nodes,
+                   Collection<PartitionInfo> partitions,
+                   Set<String> unauthorizedTopics,
+                   Set<String> invalidTopics,
+                   Set<String> internalTopics,
+                   Node controller,
+                   Map<String, Uuid> topicIds) {
+        this(clusterId, false, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds);
     }
 
     private Cluster(String clusterId,
@@ -96,7 +113,8 @@ private Cluster(String clusterId,
                     Set<String> unauthorizedTopics,
                     Set<String> invalidTopics,
                     Set<String> internalTopics,
-                    Node controller) {
+                    Node controller,
+                    Map<String, Uuid> topicIds) {
         this.isBootstrapConfigured = isBootstrapConfigured;
         this.clusterResource = new ClusterResource(clusterId);
         // make a randomized, unmodifiable copy of the nodes
@@ -165,6 +183,7 @@ private Cluster(String clusterId,
         this.partitionsByTopic = Collections.unmodifiableMap(tmpPartitionsByTopic);
         this.availablePartitionsByTopic = Collections.unmodifiableMap(tmpAvailablePartitionsByTopic);
         this.partitionsByNode = Collections.unmodifiableMap(tmpPartitionsByNode);
+        this.topicIds = Collections.unmodifiableMap(topicIds);
 
         this.unauthorizedTopics = Collections.unmodifiableSet(unauthorizedTopics);
         this.invalidTopics = Collections.unmodifiableSet(invalidTopics);
@@ -191,7 +210,7 @@ public static Cluster bootstrap(List<InetSocketAddress> addresses) {
         for (InetSocketAddress address : addresses)
             nodes.add(new Node(nodeId--, address.getHostString(), address.getPort()));
         return new Cluster(null, true, nodes, new ArrayList<>(0),
-            Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null);
+            Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());
     }
 
     /**
@@ -327,6 +346,14 @@ public Node controller() {
         return controller;
     }
 
+    public Collection<Uuid> topicIds() {
+        return topicIds.values();
+    }
+
+    public Uuid topicId(String topic) {
+        return topicIds.getOrDefault(topic, Uuid.ZERO_UUID);
+    }
+
     @Override
     public String toString() {
         return "Cluster(id = " + clusterResource.clusterId() + ", nodes = " + this.nodes +

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

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBroker;
 import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition;
@@ -125,18 +126,22 @@ public Set<String> topicsByError(Errors error) {
     public Cluster cluster() {
         Set<String> internalTopics = new HashSet<>();
         List<PartitionInfo> partitions = new ArrayList<>();
+        Map<String, Uuid> topicIds = new HashMap<>();
 
         for (TopicMetadata metadata : topicMetadata()) {
             if (metadata.error == Errors.NONE) {
                 if (metadata.isInternal)
                     internalTopics.add(metadata.topic);
+                if (metadata.topicId() != null && metadata.topicId() != Uuid.ZERO_UUID) {
+                    topicIds.put(metadata.topic, metadata.topicId());
+                }
                 for (PartitionMetadata partitionMetadata : metadata.partitionMetadata) {
                     partitions.add(toPartitionInfo(partitionMetadata, holder().brokers));
                 }
             }
         }
         return new Cluster(data.clusterId(), brokers(), partitions, topicsByError(Errors.TOPIC_AUTHORIZATION_FAILED),
-                topicsByError(Errors.INVALID_TOPIC_EXCEPTION), internalTopics, controller());
+                topicsByError(Errors.INVALID_TOPIC_EXCEPTION), internalTopics, controller(), topicIds);
     }
 
     public static PartitionInfo toPartitionInfo(PartitionMetadata metadata, Map<Integer, Node> nodesById) {
@@ -251,17 +256,20 @@ public static MetadataResponse parse(ByteBuffer buffer, short version) {
     public static class TopicMetadata {
         private final Errors error;
         private final String topic;
+        private final Uuid topicId;
         private final boolean isInternal;
         private final List<PartitionMetadata> partitionMetadata;
         private int authorizedOperations;
 
         public TopicMetadata(Errors error,
                              String topic,
+                             Uuid topicId,
                              boolean isInternal,
                              List<PartitionMetadata> partitionMetadata,
                              int authorizedOperations) {
             this.error = error;
             this.topic = topic;
+            this.topicId = topicId;
             this.isInternal = isInternal;
             this.partitionMetadata = partitionMetadata;
             this.authorizedOperations = authorizedOperations;
@@ -271,7 +279,7 @@ public TopicMetadata(Errors error,
                              String topic,
                              boolean isInternal,
                              List<PartitionMetadata> partitionMetadata) {
-            this(error, topic, isInternal, partitionMetadata, AUTHORIZED_OPERATIONS_OMITTED);
+            this(error, topic, Uuid.ZERO_UUID, isInternal, partitionMetadata, AUTHORIZED_OPERATIONS_OMITTED);
         }
 
         public Errors error() {
@@ -282,6 +290,10 @@ public String topic() {
             return topic;
         }
 
+        public Uuid topicId() {
+            return topicId;
+        }
+
         public boolean isInternal() {
             return isInternal;
         }
@@ -306,6 +318,7 @@ public boolean equals(final Object o) {
             return isInternal == that.isInternal &&
                 error == that.error &&
                 Objects.equals(topic, that.topic) &&
+                Objects.equals(topicId, that.topicId) &&
                 Objects.equals(partitionMetadata, that.partitionMetadata) &&
                 Objects.equals(authorizedOperations, that.authorizedOperations);
         }
@@ -320,6 +333,7 @@ public String toString() {
             return "TopicMetadata{" +
                 "error=" + error +
                 ", topic='" + topic + '\'' +
+                ", topicId='" + topicId + '\'' +
                 ", isInternal=" + isInternal +
                 ", partitionMetadata=" + partitionMetadata +
                 ", authorizedOperations=" + authorizedOperations +
@@ -405,6 +419,7 @@ private Collection<TopicMetadata> createTopicMetadata(MetadataResponseData data)
             for (MetadataResponseTopic topicMetadata : data.topics()) {
                 Errors topicError = Errors.forCode(topicMetadata.errorCode());
                 String topic = topicMetadata.name();
+                Uuid topicId = topicMetadata.topicId();
                 boolean isInternal = topicMetadata.isInternal();
                 List<PartitionMetadata> partitionMetadataList = new ArrayList<>();
 
@@ -422,7 +437,7 @@ private Collection<TopicMetadata> createTopicMetadata(MetadataResponseData data)
                             partitionMetadata.offlineReplicas()));
                 }
 
-                topicMetadataList.add(new TopicMetadata(topicError, topic, isInternal, partitionMetadataList,
+                topicMetadataList.add(new TopicMetadata(topicError, topic, topicId, isInternal, partitionMetadataList,
                         topicMetadata.topicAuthorizedOperations()));
             }
             return topicMetadataList;

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.UpdateMetadataRequestData;
 import org.apache.kafka.common.message.UpdateMetadataRequestData.UpdateMetadataBroker;
@@ -33,6 +34,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -44,12 +46,15 @@ public class UpdateMetadataRequest extends AbstractControlRequest {
     public static class Builder extends AbstractControlRequest.Builder<UpdateMetadataRequest> {
         private final List<UpdateMetadataPartitionState> partitionStates;
         private final List<UpdateMetadataBroker> liveBrokers;
+        private final Map<String, Uuid> topicIds;
 
         public Builder(short version, int controllerId, int controllerEpoch, long brokerEpoch,
-                       List<UpdateMetadataPartitionState> partitionStates, List<UpdateMetadataBroker> liveBrokers) {
+                       List<UpdateMetadataPartitionState> partitionStates, List<UpdateMetadataBroker> liveBrokers,
+                       Map<String, Uuid> topicIds) {
             super(ApiKeys.UPDATE_METADATA, version, controllerId, controllerEpoch, brokerEpoch);
             this.partitionStates = partitionStates;
             this.liveBrokers = liveBrokers;
+            this.topicIds = topicIds;
         }
 
         @Override
@@ -82,7 +87,7 @@ public UpdateMetadataRequest build(short version) {
                     .setLiveBrokers(liveBrokers);
 
             if (version >= 5) {
-                Map<String, UpdateMetadataTopicState> topicStatesMap = groupByTopic(partitionStates);
+                Map<String, UpdateMetadataTopicState> topicStatesMap = groupByTopic(topicIds, partitionStates);
                 data.setTopicStates(new ArrayList<>(topicStatesMap.values()));
             } else {
                 data.setUngroupedPartitionStates(partitionStates);
@@ -91,13 +96,17 @@ public UpdateMetadataRequest build(short version) {
             return new UpdateMetadataRequest(data, version);
         }
 
-        private static Map<String, UpdateMetadataTopicState> groupByTopic(List<UpdateMetadataPartitionState> partitionStates) {
+        private static Map<String, UpdateMetadataTopicState> groupByTopic(Map<String, Uuid> topicIds, List<UpdateMetadataPartitionState> partitionStates) {
             Map<String, UpdateMetadataTopicState> topicStates = new HashMap<>();
             for (UpdateMetadataPartitionState partition : partitionStates) {
                 // We don't null out the topic name in UpdateMetadataPartitionState since it's ignored by the generated
                 // code if version >= 5
                 UpdateMetadataTopicState topicState = topicStates.computeIfAbsent(partition.topicName(),
-                    t -> new UpdateMetadataTopicState().setTopicName(partition.topicName()));
+                    t -> new UpdateMetadataTopicState()
+                            .setTopicName(partition.topicName())
+                            .setTopicId(topicIds.getOrDefault(partition.topicName(), Uuid.ZERO_UUID))
+
+                );
                 topicState.partitionStates().add(partition);
             }
             return topicStates;
@@ -196,6 +205,13 @@ public Iterable<UpdateMetadataPartitionState> partitionStates() {
         return data.ungroupedPartitionStates();
     }
 
+    public List<UpdateMetadataTopicState> topicStates() {
+        if (version() >= 5) {
+            return data.topicStates();
+        }
+        return Collections.emptyList();
+    }
+
     public List<UpdateMetadataBroker> liveBrokers() {
         return data.liveBrokers();
     }

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

@@ -17,7 +17,7 @@
   "apiKey": 3,
   "type": "request",
   "name": "MetadataRequest",
-  "validVersions": "0-9",
+  "validVersions": "0-10",
   "flexibleVersions": "9+",
   "fields": [
     // In version 0, an empty array indicates "request metadata for all topics."  In version 1 and
@@ -31,9 +31,11 @@
     // Starting in version 8, authorized operations can be requested for cluster and topic resource.
     //
     // Version 9 is the first flexible version.
+    // Version 10 add topicId
     { "name": "Topics", "type": "[]MetadataRequestTopic", "versions": "0+", "nullableVersions": "1+",
       "about": "The topics to fetch metadata for.", "fields": [
-      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+      { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true, "about": "The topic id." },
+      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", "nullableVersions": "10+",
         "about": "The topic name." }
     ]},
     { "name": "AllowAutoTopicCreation", "type": "bool", "versions": "4+", "default": "true", "ignorable": false,

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

@@ -36,7 +36,8 @@
   // Starting in version 8, brokers can send authorized operations for topic and cluster.
   //
   // Version 9 is the first flexible version.
-  "validVersions": "0-9",
+  // Version 10 add topicId
+  "validVersions": "0-10",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "3+", "ignorable": true,
@@ -62,6 +63,7 @@
         "about": "The topic error, or 0 if there was no error." },
       { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
         "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true, "about": "The topic id." },
       { "name": "IsInternal", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
         "about": "True if the topic is internal." },
       { "name": "Partitions", "type": "[]MetadataResponsePartition", "versions": "0+",

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

@@ -26,7 +26,8 @@
   // Version 4 adds the offline replica list.
   //
   // Version 5 adds the broker epoch field and normalizes partitions by topic.
-  "validVersions": "0-6",
+  // Version 7 adds topicId
+  "validVersions": "0-7",
   "flexibleVersions": "6+",
   "fields": [
     { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -41,6 +42,7 @@
       "about": "In newer versions of this RPC, each topic that we would like to update.", "fields": [
       { "name": "TopicName", "type": "string", "versions": "5+", "entityType": "topicName",
         "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "7+", "ignorable": true, "about": "The topic id."},
       { "name": "PartitionStates", "type": "[]UpdateMetadataPartitionState", "versions": "5+",
         "about": "The partition that we would like to update." }
     ]},

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

@@ -18,7 +18,7 @@
   "type": "response",
   "name": "UpdateMetadataResponse",
   // Versions 1, 2, 3, 4, and 5 are the same as version 0
-  "validVersions": "0-6",
+  "validVersions": "0-7",
   "flexibleVersions": "6+",
   "fields": [
       { "name": "ErrorCode", "type": "int16", "versions": "0+",