KIP-602: Support client_dns_lookup in Consumer/Producer/Admin config (#3004)

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

Author: dpkp

kafka/admin/client.py

@@ -169,6 +169,7 @@ class KafkaAdminClient(
         'sock_chunk_buffer_count': 1000,  # undocumented experimental option
         'retry_backoff_ms': 100,
         'metadata_max_age_ms': 300000,
+        'client_dns_lookup': 'use_all_dns_ips',
         'security_protocol': 'PLAINTEXT',
         'ssl_context': None,
         'ssl_check_hostname': True,

kafka/cluster.py

@@ -44,6 +44,7 @@ class ClusterMetadata:
         'metadata_max_age_ms': 300000,
         'bootstrap_servers': [],
         'allow_auto_create_topics': True,
+        'client_dns_lookup': 'use_all_dns_ips',
     }
 
     def __init__(self, **configs):
@@ -180,6 +181,9 @@ def _generate_bootstrap_brokers(self):
         # collect_hosts does not perform DNS, so we should be fine to re-use
         bootstrap_hosts = collect_hosts(self.config['bootstrap_servers'])
 
+        if self.config['client_dns_lookup'] == 'resolve_canonical_bootstrap_servers_only':
+            bootstrap_hosts = expand_to_canonical_bootstrap_hosts(bootstrap_hosts)
+
         brokers = {}
         for i, (host, port, _) in enumerate(bootstrap_hosts):
             node_id = 'bootstrap-%s' % i
@@ -615,8 +619,9 @@ def __str__(self):
 
 def collect_hosts(hosts, randomize=True):
     """
-    Collects a comma-separated set of hosts (host:port) and optionally
-    randomize the returned list.
+    Processes a list (or comma-separated string) of hosts strings (host:port)
+    and returns a list of (host, port, family) tuples.
+    Optionally randomizes the returned list.
     """
 
     if isinstance(hosts, str):
@@ -634,6 +639,38 @@ def collect_hosts(hosts, randomize=True):
     return result
 
 
+def expand_to_canonical_bootstrap_hosts(hosts):
+    """Expand each bootstrap entry to one entry per canonical FQDN.
+
+    Mirrors Java's ``client.dns.lookup=resolve_canonical_bootstrap_servers_only``:
+    forward-resolve each host, take the ``canonname`` reported by the resolver,
+    and emit one bootstrap entry per unique canonical name. Useful for
+    Kerberos round-robin DNS deployments where the principal must match each
+    individual broker FQDN.
+
+    If a host fails to resolve, the original entry is preserved verbatim --
+    matching Java's best-effort behaviour so bootstrap doesn't fail outright.
+    """
+    expanded = []
+    for host, port, afi in hosts:
+        try:
+            addrinfos = socket.getaddrinfo(
+                host, port, afi, socket.SOCK_STREAM, 0, socket.AI_CANONNAME)
+        except socket.gaierror as exc:
+            log.warning('Canonical bootstrap resolution failed for %s:%s: %s; '
+                        'keeping original entry', host, port, exc)
+            expanded.append((host, port, afi))
+            continue
+        seen = set()
+        for family, _socktype, _proto, canonname, _sockaddr in addrinfos:
+            name = canonname or host
+            if name in seen:
+                continue
+            seen.add(name)
+            expanded.append((name, port, family))
+    return expanded
+
+
 def _address_family(address):
     """
         Attempt to determine the family of an address (or hostname)

kafka/consumer/group.py

@@ -311,6 +311,7 @@ class KafkaConsumer:
         'isolation_level': 'read_uncommitted',
         'allow_auto_create_topics': True,
         'metadata_max_age_ms': 5 * 60 * 1000,
+        'client_dns_lookup': 'use_all_dns_ips',
         'partition_assignment_strategy': (RangePartitionAssignor, RoundRobinPartitionAssignor),
         'max_poll_records': 500,
         'max_poll_interval_ms': 300000,

kafka/net/manager.py

@@ -57,13 +57,21 @@ class KafkaConnectionManager:
         'metrics': None,
         'metric_group_prefix': '',
         'metadata_max_age_ms': 300000,
+        'client_dns_lookup': 'use_all_dns_ips',
     }
+    _VALID_DNS_LOOKUP_MODES = ('use_all_dns_ips', 'resolve_canonical_bootstrap_servers_only')
+
     def __init__(self, net, **configs):
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
             if key in configs:
                 self.config[key] = configs[key]
 
+        if self.config['client_dns_lookup'] not in self._VALID_DNS_LOOKUP_MODES:
+            raise ValueError(
+                "client_dns_lookup must be one of %s; got %r"
+                % (self._VALID_DNS_LOOKUP_MODES, self.config['client_dns_lookup']))
+
         if 'socks5_proxy' in configs:
             if self.config['proxy_url'] is None:
                 log.warning('socks5_proxy is deprecated, use proxy_url instead')
@@ -73,6 +81,7 @@ def __init__(self, net, **configs):
         self.cluster = ClusterMetadata(
             bootstrap_servers=self.config['bootstrap_servers'],
             metadata_max_age_ms=self.config['metadata_max_age_ms'],
+            client_dns_lookup=self.config['client_dns_lookup'],
         )
         self.cluster.attach(self)
         self._conns = {}

kafka/producer/kafka.py

@@ -398,6 +398,7 @@ class KafkaProducer:
         'max_request_size': 1048576,
         'allow_auto_create_topics': True,
         'metadata_max_age_ms': 300000,
+        'client_dns_lookup': 'use_all_dns_ips',
         'retry_backoff_ms': 100,
         'request_timeout_ms': 30000,
         'receive_buffer_bytes': None,

test/net/test_manager.py

@@ -42,6 +42,21 @@ def test_api_versions(self, net):
         m = KafkaConnectionManager(net, api_version=(1, 0))
         assert m.broker_version_data == BrokerVersionData((1, 0))
 
+    def test_client_dns_lookup_default(self, net):
+        m = KafkaConnectionManager(net)
+        assert m.config['client_dns_lookup'] == 'use_all_dns_ips'
+        assert m.cluster.config['client_dns_lookup'] == 'use_all_dns_ips'
+
+    def test_client_dns_lookup_canonical_passthrough(self, net):
+        m = KafkaConnectionManager(net, client_dns_lookup='resolve_canonical_bootstrap_servers_only')
+        assert m.cluster.config['client_dns_lookup'] == 'resolve_canonical_bootstrap_servers_only'
+
+    def test_client_dns_lookup_invalid_rejected(self, net):
+        with pytest.raises(ValueError, match='client_dns_lookup'):
+            KafkaConnectionManager(net, client_dns_lookup='default')
+        with pytest.raises(ValueError, match='client_dns_lookup'):
+            KafkaConnectionManager(net, client_dns_lookup='garbage')
+
 
 class TestKafkaConnectionManagerProxyConfig:
     def test_proxy_url_default_none(self, net):

test/test_cluster.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from kafka.cluster import collect_hosts
+from kafka.cluster import ClusterMetadata, collect_hosts, expand_to_canonical_bootstrap_hosts
 from kafka.future import Future
 from kafka.protocol.metadata import MetadataResponse
 from kafka.structs import TopicPartition
@@ -241,6 +241,64 @@ def test_collect_hosts__protocol(self):
         ])
 
 
+class TestExpandToCanonicalBootstrapHosts:
+    def test_expands_multi_ip_host_to_canonical_names(self):
+        addrinfos = [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-1.kafka.example.com', ('10.0.0.1', 9092)),
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-2.kafka.example.com', ('10.0.0.2', 9092)),
+        ]
+        with patch('socket.getaddrinfo', return_value=addrinfos):
+            expanded = expand_to_canonical_bootstrap_hosts([('kafka.example.com', 9092, socket.AF_UNSPEC)])
+        assert expanded == [
+            ('broker-1.kafka.example.com', 9092, socket.AF_INET),
+            ('broker-2.kafka.example.com', 9092, socket.AF_INET),
+        ]
+
+    def test_deduplicates_canonical_names(self):
+        addrinfos = [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-1.kafka.example.com', ('10.0.0.1', 9092)),
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-1.kafka.example.com', ('10.0.0.5', 9092)),
+        ]
+        with patch('socket.getaddrinfo', return_value=addrinfos):
+            expanded = expand_to_canonical_bootstrap_hosts([('kafka.example.com', 9092, socket.AF_UNSPEC)])
+        assert expanded == [('broker-1.kafka.example.com', 9092, socket.AF_INET)]
+
+    def test_falls_back_to_original_on_resolution_failure(self):
+        with patch('socket.getaddrinfo', side_effect=socket.gaierror('no such host')):
+            expanded = expand_to_canonical_bootstrap_hosts([('kafka.example.com', 9092, socket.AF_UNSPEC)])
+        assert expanded == [('kafka.example.com', 9092, socket.AF_UNSPEC)]
+
+    def test_missing_canonname_falls_back_to_input_host(self):
+        addrinfos = [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, '', ('10.0.0.1', 9092)),
+        ]
+        with patch('socket.getaddrinfo', return_value=addrinfos):
+            expanded = expand_to_canonical_bootstrap_hosts([('kafka.example.com', 9092, socket.AF_UNSPEC)])
+        assert expanded == [('kafka.example.com', 9092, socket.AF_INET)]
+
+
+class TestClusterMetadataClientDnsLookup:
+    def test_default_does_not_expand_bootstrap(self):
+        with patch('socket.getaddrinfo') as mock_gai:
+            cluster = ClusterMetadata(bootstrap_servers='kafka.example.com:9092')
+            assert mock_gai.call_count == 0
+        nodes = cluster.bootstrap_brokers()
+        assert [n.host for n in nodes] == ['kafka.example.com']
+
+    def test_canonical_mode_expands_bootstrap(self):
+        addrinfos = [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-1.kafka.example.com', ('10.0.0.1', 9092)),
+            (socket.AF_INET, socket.SOCK_STREAM, 0, 'broker-2.kafka.example.com', ('10.0.0.2', 9092)),
+        ]
+        with patch('socket.getaddrinfo', return_value=addrinfos):
+            cluster = ClusterMetadata(
+                bootstrap_servers='kafka.example.com:9092',
+                client_dns_lookup='resolve_canonical_bootstrap_servers_only')
+        nodes = cluster.bootstrap_brokers()
+        assert sorted(n.host for n in nodes) == [
+            'broker-1.kafka.example.com', 'broker-2.kafka.example.com']
+
+
 class TestClusterMetadataRefresh:
     def test_request_update_returns_future(self, cluster):
         f = cluster.request_update()