KIP-516: Initial topic id (uuid) support (#3031)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dpkp

kafka/admin/_cluster.py

@@ -6,6 +6,7 @@
 from enum import IntEnum
 import logging
 from typing import TYPE_CHECKING
+import uuid
 
 import kafka.errors as Errors
 from kafka.protocol.api_key import ApiKey
@@ -30,11 +31,20 @@ class ClusterAdminMixin:
     _manager: KafkaConnectionManager
 
     async def _get_cluster_metadata(self, topics):
-        """topics = [] for no topics, None for all."""
+        """topics = [] for no topics, None for all. Items may be topic-name
+        strings or :class:`uuid.UUID` topic ids (KIP-516, requires broker
+        >= 2.8 / MetadataRequest v12+)."""
+        _Topic = MetadataRequest.MetadataRequestTopic
+        if topics is None:
+            request_topics = None
+        else:
+            request_topics = [
+                _Topic(name=None, topic_id=t) if isinstance(t, uuid.UUID)
+                else _Topic(name=t)
+                for t in topics
+            ]
         request = MetadataRequest(
-            topics=[
-                MetadataRequest.MetadataRequestTopic(name=topic)
-                for topic in topics] if topics is not None else None,
+            topics=request_topics,
             allow_auto_topic_creation=False,
             include_cluster_authorized_operations=True,
             include_topic_authorized_operations=True,

kafka/admin/_topics.py

@@ -38,8 +38,11 @@ def describe_topics(self, topics=None):
         """Fetch metadata for the specified topics or all topics if None.
 
         Keyword Arguments:
-            topics ([str], optional) A list of topic names. If None, metadata for all
-                topics is retrieved.
+            topics (list, optional): A list of topic names or
+                :class:`uuid.UUID` topic ids (KIP-516). Strings and UUIDs may
+                be mixed. Describing by id requires broker >= 2.8
+                (MetadataRequest v12+); name-based describe works on any
+                broker. If None, metadata for all topics is retrieved.
 
         Returns:
             A list of dicts describing each topic (including partition info).
@@ -110,7 +113,7 @@ def create_topics(self, new_topics, timeout_ms=None, validate_only=False, raise_
                 in broker metadata with a leader assigned for every partition. Default: False
 
         Returns:
-            dict of CreateTopicResponse key/vals
+            dict of CreateTopicsResponse key/vals.
         """
         if validate_only and wait_for_metadata:
             raise ValueError('validate_only and wait_for_metadata are mutually exclusive')
@@ -131,7 +134,6 @@ def create_topics(self, new_topics, timeout_ms=None, validate_only=False, raise_
             topics=topics,
             timeout_ms=timeout_ms,
             validate_only=validate_only,
-            max_version=3,
         )
         def response_errors(r):
             for topic in r.topics:

kafka/cli/admin/topics/describe.py

@@ -1,11 +1,18 @@
+import uuid
+
+
 class DescribeTopics:
     COMMAND = 'describe'
     HELP = 'Describe Kafka Topics'
 
     @classmethod
     def add_arguments(cls, parser):
-        parser.add_argument('-t', '--topic', type=str, action='append', dest='topics')
+        parser.add_argument('-t', '--topic', type=str, action='append', dest='topics', default=[], help='topic name')
+        parser.add_argument('--id', type=str, action='append', dest='topic_ids', default=[],
+                            help='topic UUID (requires broker >= 2.8, KIP-516)')
 
     @classmethod
     def command(cls, client, args):
-        return client.describe_topics(args.topics or None)
+        topic_ids = [uuid.UUID(topic_id) for topic_id in args.topic_ids]
+        topics = args.topics + topic_ids
+        return client.describe_topics(topics or None)

kafka/cluster.py

@@ -6,6 +6,7 @@
 import socket
 import threading
 import time
+import uuid
 import weakref
 
 from kafka import errors as Errors
@@ -53,6 +54,8 @@ def __init__(self, **configs):
         self._brokers = {}  # node_id -> MetadataResponseBroker
         self._partitions = {}  # topic -> partition -> PartitionMetadata
         self._broker_partitions = collections.defaultdict(set)  # node_id -> {TopicPartition...}
+        self._topic_ids = {}  # topic name -> uuid.UUID
+        self._topic_names_by_id = {}  # uuid.UUID -> topic name
         self._coordinators = {}  # (key_type, key) -> node_id
         self._last_refresh_ms = 0
         self._last_successful_refresh_ms = 0
@@ -472,6 +475,22 @@ def metadata_request(self):
             include_topic_authorized_operations=False,
         )
 
+    def topic_id(self, topic_name):
+        """Return the topic UUID for ``topic_name``, or None if unknown.
+
+        Populated from MetadataResponse v10+ (Kafka 2.8+, KIP-516). Older
+        responses leave this empty.
+        """
+        return self._topic_ids.get(topic_name)
+
+    def topic_name_for_id(self, topic_id):
+        """Return the topic name for ``topic_id`` (uuid.UUID), or None.
+
+        Reverse lookup of :meth:`topic_id`. Populated from MetadataResponse
+        v10+ (KIP-516).
+        """
+        return self._topic_names_by_id.get(topic_id)
+
     def failed_update(self, exception):
         """Update cluster state given a failed MetadataRequest."""
         f = None
@@ -520,6 +539,8 @@ def update_metadata(self, metadata):
         _new_broker_partitions = collections.defaultdict(set)
         _new_unauthorized_topics = set()
         _new_internal_topics = set()
+        _new_topic_ids = {}
+        _new_topic_names_by_id = {}
         _retry_topics = set()
 
         # KAFKA-9212: pre-2.4 brokers may emit stale leader_epoch values
@@ -537,11 +558,23 @@ def update_metadata(self, metadata):
             if t.is_internal:
                 _new_internal_topics.add(topic)
             error_type = Errors.for_code(t.error_code)
+            new_topic_id = t.topic_id
+            recreated = False
+            if new_topic_id is not None and topic is not None:
+                prior = self._topic_ids.get(topic)
+                if prior is not None and prior != new_topic_id:
+                    log.warning(
+                        "Topic %s topic_id changed from %s to %s -- likely"
+                        " recreated; resetting cached leader epochs.",
+                        topic, prior, new_topic_id)
+                    recreated = True
+                _new_topic_ids[topic] = new_topic_id
+                _new_topic_names_by_id[new_topic_id] = topic
             if error_type is Errors.NoError:
                 _new_partitions[topic] = {}
                 for p_data in t.partitions:
                     partition = p_data.partition_index
-                    if not epoch_reliable:
+                    if not epoch_reliable or recreated:
                         p_data.leader_epoch = -1
                     _new_partitions[topic][partition] = p_data
                     if p_data.leader_id != -1:
@@ -576,6 +609,13 @@ def update_metadata(self, metadata):
             self._broker_partitions = _new_broker_partitions
             self.unauthorized_topics = _new_unauthorized_topics
             self.internal_topics = _new_internal_topics
+            # Pre-v10 responses don't carry topic_id, so the wholesale swap
+            # would clobber known ids during a rolling downgrade (or any
+            # cross-broker version skew). Only replace the index when the
+            # response actually had a chance to populate it.
+            if metadata.API_VERSION >= 10:
+                self._topic_ids = _new_topic_ids
+                self._topic_names_by_id = _new_topic_names_by_id
             self._need_update = len(_retry_topics) > 0
             f = None
             if self._future:

test/admin/test_admin_topics.py

@@ -1,7 +1,16 @@
+import uuid
+
 import pytest
 
 from kafka.admin import KafkaAdminClient, NewTopic, NewPartitions
 from kafka.errors import KafkaTimeoutError, UnknownTopicOrPartitionError
+from kafka.protocol.admin import (
+    CreateTopicsRequest,
+    CreateTopicsResponse,
+    DeleteTopicsRequest,
+    DeleteTopicsResponse,
+)
+from kafka.protocol.metadata import MetadataRequest, MetadataResponse
 
 
 def test_new_partitions():
@@ -190,3 +199,112 @@ def fake_describe_topics(topics):
     monkeypatch.setattr(admin, 'describe_topics', fake_describe_topics)
     admin.wait_for_topics(['foo'], timeout_ms=5000)
     assert state['calls'] == 2
+
+
+# ---------------------------------------------------------------------------
+# KIP-516: topic UUIDs in admin requests/responses
+# ---------------------------------------------------------------------------
+
+
+class TestTopicIdAdmin:
+    """KIP-516 topic ids in the admin surface: delete-by-id, describe-by-id,
+    and CreateTopics surfacing topic_id from v7+ responses."""
+
+    def test_delete_topics_by_id_encodes_topic_id(self, broker, admin):
+        """delete_topics already accepts uuid.UUID items; pin the wire shape."""
+        captured = {}
+
+        def handler(api_key, api_version, correlation_id, request_bytes):
+            decoded = DeleteTopicsRequest.decode(
+                request_bytes, version=api_version, header=True)
+            captured['request'] = decoded
+            captured['version'] = api_version
+            return DeleteTopicsResponse(throttle_time_ms=0, responses=[])
+
+        broker.respond_fn(DeleteTopicsRequest, handler)
+
+        u = uuid.uuid4()
+        admin.delete_topics([u])
+
+        req = captured['request']
+        # v6+ uses the structured DeleteTopicState carrying topic_id.
+        assert captured['version'] >= 6
+        assert len(req.topics) == 1
+        assert req.topics[0].topic_id == u
+        assert req.topics[0].name is None
+
+    def test_describe_topics_by_id_sends_topic_id(self, broker, admin):
+        captured = {}
+
+        def handler(api_key, api_version, correlation_id, request_bytes):
+            decoded = MetadataRequest.decode(
+                request_bytes, version=api_version, header=True)
+            captured['request'] = decoded
+            captured['version'] = api_version
+            return MetadataResponse(throttle_time_ms=0, brokers=[], cluster_id='c',
+                                    controller_id=0, topics=[])
+
+        broker.respond_fn(MetadataRequest, handler)
+
+        u = uuid.uuid4()
+        admin.describe_topics([u])
+
+        req = captured['request']
+        # MetadataRequest must be v12+ for name=null + topic_id to work.
+        assert captured['version'] >= 12
+        assert len(req.topics) == 1
+        assert req.topics[0].topic_id == u
+        assert req.topics[0].name is None
+
+    def test_describe_topics_mixed_name_and_id(self, broker, admin):
+        captured = {}
+
+        def handler(api_key, api_version, correlation_id, request_bytes):
+            decoded = MetadataRequest.decode(
+                request_bytes, version=api_version, header=True)
+            captured['request'] = decoded
+            return MetadataResponse(throttle_time_ms=0, brokers=[], cluster_id='c',
+                                    controller_id=0, topics=[])
+
+        broker.respond_fn(MetadataRequest, handler)
+
+        u = uuid.uuid4()
+        admin.describe_topics(['by-name', u])
+
+        req = captured['request']
+        assert len(req.topics) == 2
+        assert req.topics[0].name == 'by-name'
+        assert req.topics[0].topic_id is None
+        assert req.topics[1].name is None
+        assert req.topics[1].topic_id == u
+
+    def test_create_topics_negotiates_v7_and_surfaces_topic_id(self, broker, admin):
+        """With max_version=3 dropped, negotiation should reach v7+ on a
+        modern MockBroker and the response's topic_id flows through to_dict()."""
+        captured = {}
+        u = uuid.uuid4()
+
+        def handler(api_key, api_version, correlation_id, request_bytes):
+            captured['version'] = api_version
+            _Result = CreateTopicsResponse.CreatableTopicResult
+            return CreateTopicsResponse(
+                throttle_time_ms=0,
+                topics=[_Result(
+                    name='foo',
+                    topic_id=u,
+                    error_code=0,
+                    error_message=None,
+                    num_partitions=1,
+                    replication_factor=1,
+                    configs=[],
+                )],
+            )
+
+        broker.respond_fn(CreateTopicsRequest, handler)
+
+        result = admin.create_topics({'foo': {'num_partitions': 1, 'replication_factor': 1}})
+
+        # Default MockBroker is (4, 2); negotiation should land on v7+.
+        assert captured['version'] >= 7
+        # to_dict() stringifies the UUID; compare via uuid.UUID round-trip.
+        assert uuid.UUID(result['topics'][0]['topic_id']) == u