Move kafka.sasl -> kafka.net.sasl (#3056)

dpkp · web-flow · commit e394544ff832 · 2026-06-10T19:20:38.000-07:00
diff --git a/CHANGES.md b/CHANGES.md
@@ -36,6 +36,9 @@ This is a major release with significant changes to kafka-python internals to si
 * Remove kafka.client_async / kafka.conn legacy modules (#2918)
 * Drop version probes for pre-0.10/ApiVersionsRequest brokers
 
+### SASL Module Rename
+* Rename kafka.sasl -> kafka.net.sasl
+
 ## Networking (kafka.net)
 
 Complete refactor of the networking layer using a bespoke event-loop supporting async/await (but no asyncio yet). All three clients (Admin, Consumer, Producer) use a dedicated IO thread that drives a selector-based event loop. AdminClient/KafkaConsumer leverage a built-in io thread supplied by kafka.net, KafkaProducer continues to use its existing background Sender thread for now.
diff --git a/docs/changelog.rst b/docs/changelog.rst
@@ -48,6 +48,10 @@ Old Networking Stack Removal
 * Remove kafka.client_async / kafka.conn legacy modules (#2918)
 * Drop version probes for pre-0.10/ApiVersionsRequest brokers
 
+SASL Module Rename
+^^^^^^^^^^^^^^^^^^
+* Rename kafka.sasl -> kafka.net.sasl
+
 Networking (kafka.net)
 ----------------------
 
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
@@ -174,7 +174,7 @@ class KafkaAdminClient(
             sasl mechanism handshake. Default: 'kafka'
         sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
             sasl mechanism handshake. Default: one of bootstrap servers
-        sasl_oauth_token_provider (kafka.sasl.oauth.AbstractTokenProvider): OAuthBearer
+        sasl_oauth_token_provider (kafka.net.sasl.oauth.AbstractTokenProvider): OAuthBearer
             token provider instance. Default: None
         proxy_url (str): URL to proxy socket connections through. Supports SOCKS5 only.
             Requires scheme:// (e.g., socks5://foo.bar/). Default: None
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
@@ -283,7 +283,7 @@ class KafkaConsumer:
             sasl mechanism handshake. Default: 'kafka'
         sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
             sasl mechanism handshake. Default: one of bootstrap servers
-        sasl_oauth_token_provider (kafka.sasl.oauth.AbstractTokenProvider): OAuthBearer
+        sasl_oauth_token_provider (kafka.net.sasl.oauth.AbstractTokenProvider): OAuthBearer
             token provider instance. Default: None
         proxy_url (str): URL to proxy socket connections through. Supports SOCKS5 only.
             Requires scheme:// (e.g., socks5://foo.bar/). Default: None
diff --git a/kafka/net/connection.py b/kafka/net/connection.py
@@ -8,11 +8,11 @@
 import kafka.errors as Errors
 from kafka.future import Future
 from kafka.net.metrics import KafkaConnectionMetrics
+from kafka.net.sasl import get_sasl_mechanism
 from kafka.protocol.metadata import ApiVersionsRequest
 from kafka.protocol.sasl import SaslAuthenticateRequest, SaslHandshakeRequest, SaslBytesRequest
 from kafka.protocol.broker_version_data import BrokerVersionData
 from kafka.protocol.parser import KafkaProtocol
-from kafka.sasl import get_sasl_mechanism
 from kafka.version import __version__
 
 
diff --git a/kafka/net/sasl/__init__.py b/kafka/net/sasl/__init__.py
@@ -1,11 +1,11 @@
 import platform
 
-from kafka.sasl.gssapi import SaslMechanismGSSAPI
-from kafka.sasl.msk import SaslMechanismAwsMskIam
-from kafka.sasl.oauth import SaslMechanismOAuth
-from kafka.sasl.plain import SaslMechanismPlain
-from kafka.sasl.scram import SaslMechanismScram
-from kafka.sasl.sspi import SaslMechanismSSPI
+from .gssapi import SaslMechanismGSSAPI
+from .msk import SaslMechanismAwsMskIam
+from .oauth import SaslMechanismOAuth
+from .plain import SaslMechanismPlain
+from .scram import SaslMechanismScram
+from .sspi import SaslMechanismSSPI
 
 
 SASL_MECHANISMS = {}
diff --git a/kafka/net/sasl/abc.py b/kafka/net/sasl/abc.py
diff --git a/kafka/net/sasl/gssapi.py b/kafka/net/sasl/gssapi.py
@@ -9,7 +9,7 @@
     gssapi = None
     GSSError = None
 
-from kafka.sasl.abc import SaslMechanism
+from .abc import SaslMechanism
 
 
 class SaslMechanismGSSAPI(SaslMechanism):
diff --git a/kafka/net/sasl/msk.py b/kafka/net/sasl/msk.py
@@ -13,8 +13,8 @@
     # no botocore available, will disable AWS_MSK_IAM mechanism
     BotoSession = None
 
+from .abc import SaslMechanism
 from kafka.errors import KafkaConfigurationError
-from kafka.sasl.abc import SaslMechanism
 
 
 log = logging.getLogger(__name__)
diff --git a/kafka/net/sasl/oauth.py b/kafka/net/sasl/oauth.py
@@ -1,8 +1,8 @@
 from abc import ABC, abstractmethod
 import logging
 
+from .abc import SaslMechanism
 from kafka.errors import KafkaConfigurationError
-from kafka.sasl.abc import SaslMechanism
 
 
 log = logging.getLogger(__name__)
@@ -14,7 +14,7 @@ def __init__(self, **config):
         if 'sasl_oauth_token_provider' not in config:
             raise KafkaConfigurationError('sasl_oauth_token_provider required for OAUTHBEARER sasl')
         if not isinstance(config['sasl_oauth_token_provider'], AbstractTokenProvider):
-            raise KafkaConfigurationError('sasl_oauth_token_provider must implement kafka.sasl.oauth.AbstractTokenProvider')
+            raise KafkaConfigurationError('sasl_oauth_token_provider must implement kafka.net.sasl.oauth.AbstractTokenProvider')
         self.token_provider = config['sasl_oauth_token_provider']
         self._error = None
         self._is_done = False
diff --git a/kafka/net/sasl/plain.py b/kafka/net/sasl/plain.py
@@ -1,7 +1,7 @@
 import logging
 
+from .abc import SaslMechanism
 from kafka.errors import KafkaConfigurationError
-from kafka.sasl.abc import SaslMechanism
 
 
 log = logging.getLogger(__name__)
diff --git a/kafka/net/sasl/scram.py b/kafka/net/sasl/scram.py
@@ -5,8 +5,8 @@
 import uuid
 
 
+from .abc import SaslMechanism
 from kafka.errors import KafkaConfigurationError
-from kafka.sasl.abc import SaslMechanism
 
 
 log = logging.getLogger(__name__)
diff --git a/kafka/net/sasl/sspi.py b/kafka/net/sasl/sspi.py
@@ -9,8 +9,8 @@
 except ImportError:
     sspi = None
 
+from .abc import SaslMechanism
 from kafka.errors import KafkaConfigurationError
-from kafka.sasl.abc import SaslMechanism
 
 
 log = logging.getLogger(__name__)
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
@@ -391,7 +391,7 @@ class KafkaProducer:
             sasl mechanism handshake. Default: 'kafka'
         sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI
             sasl mechanism handshake. Default: one of bootstrap servers
-        sasl_oauth_token_provider (kafka.sasl.oauth.AbstractTokenProvider): OAuthBearer
+        sasl_oauth_token_provider (kafka.net.sasl.oauth.AbstractTokenProvider): OAuthBearer
             token provider instance. Default: None
         proxy_url (str): URL to proxy socket connections through. Supports SOCKS5 only.
             Requires scheme:// (e.g., socks5://foo.bar/). Default: None
diff --git a/test/net/sasl/test_gssapi.py b/test/net/sasl/test_gssapi.py
@@ -1,7 +1,7 @@
 from unittest import mock
 
-from kafka.sasl import get_sasl_mechanism
-import kafka.sasl.gssapi
+from kafka.net.sasl import get_sasl_mechanism
+import kafka.net.sasl.gssapi
 
 
 def test_gssapi():
@@ -18,10 +18,10 @@ def mocked_message_wrapper(msg, *args):
         return wrapped
     client_ctx.unwrap.side_effect = mocked_message_wrapper
     client_ctx.wrap.side_effect = mocked_message_wrapper
-    kafka.sasl.gssapi.gssapi = mock.Mock()
-    kafka.sasl.gssapi.gssapi.SecurityContext.return_value = client_ctx
+    kafka.net.sasl.gssapi.gssapi = mock.Mock()
+    kafka.net.sasl.gssapi.gssapi.SecurityContext.return_value = client_ctx
     gssapi = get_sasl_mechanism('GSSAPI')(**config)
-    assert isinstance(gssapi, kafka.sasl.gssapi.SaslMechanismGSSAPI)
+    assert isinstance(gssapi, kafka.net.sasl.gssapi.SaslMechanismGSSAPI)
     client_ctx.step.assert_called_with(None)
 
     while not gssapi.is_done():
diff --git a/test/net/sasl/test_msk.py b/test/net/sasl/test_msk.py
@@ -3,15 +3,15 @@
 import sys
 from unittest import mock
 
-from kafka.sasl.msk import AwsMskIamClient, SaslMechanismAwsMskIam
+from kafka.net.sasl.msk import AwsMskIamClient, SaslMechanismAwsMskIam
 
 
 def client_factory(token=None):
     if sys.version_info >= (3, 3):
         now = datetime.datetime.fromtimestamp(1629321911, datetime.timezone.utc)
     else:
         now = datetime.datetime.utcfromtimestamp(1629321911)
-    with mock.patch('kafka.sasl.msk.datetime') as mock_dt:
+    with mock.patch('kafka.net.sasl.msk.datetime') as mock_dt:
         mock_dt.datetime.utcnow = mock.Mock(return_value=now)
         return AwsMskIamClient(
             host='localhost',
@@ -68,7 +68,7 @@ def test_aws_msk_iam_client_temporary_credentials():
 
 
 def test_aws_msk_iam_sasl_mechanism():
-    with mock.patch('kafka.sasl.msk.BotoSession'):
+    with mock.patch('kafka.net.sasl.msk.BotoSession'):
         sasl = SaslMechanismAwsMskIam(security_protocol='SASL_SSL', host='localhost')
         with mock.patch.object(sasl, '_build_client', return_value=client_factory(token=None)):
             assert sasl.auth_bytes() != b''
diff --git a/test/net/test_connection.py b/test/net/test_connection.py
@@ -597,8 +597,8 @@ def mock_send_request(request, **kwargs):
         conn._send_request = mock_send_request
 
         captured = {}
-        from kafka.sasl import register_sasl_mechanism
-        from kafka.sasl.plain import SaslMechanismPlain
+        from kafka.net.sasl import register_sasl_mechanism
+        from kafka.net.sasl.plain import SaslMechanismPlain
 
         class RecordingPlain(SaslMechanismPlain):
             def __init__(self, **config):
