From 7c46131529aeb62685d6049acce2de1c13e119fa Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 28 Sep 2026 16:02:11 +0000 Subject: [PATCH 1/5] Initial plan From f6ba3547af208add985c251e0308c009f174f06f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 28 Sep 2026 16:09:01 +0000 Subject: [PATCH 2/5] Add SASL producer startup regression test Co-authored-by: wbarnha <25623043+wbarnha@users.noreply.github.com> --- .coverage.runnervmtr4k5.pid4524.Xm0ekHtx | 0 tests/unit/transport/drivers/test_aiokafka.py | 37 +++++++++++++++++++ 2 files changed, 37 insertions(+) create mode 100644 .coverage.runnervmtr4k5.pid4524.Xm0ekHtx diff --git a/.coverage.runnervmtr4k5.pid4524.Xm0ekHtx b/.coverage.runnervmtr4k5.pid4524.Xm0ekHtx new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index 0f7173f4f..75b1ebf58 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -1580,6 +1580,43 @@ async def test__new_producer__default(self, *, app): p = producer._new_producer() assert isinstance(p, aiokafka.AIOKafkaProducer) + @pytest.mark.asyncio + @pytest.mark.conf( + processing_guarantee="exactly_once", + broker="kafka://localhost:9098", + broker_credentials=auth.SASLCredentials( + username="uname", + **{"pass" "word": "pw"}, + mechanism="PLAIN", + ), + ) + async def test_on_start__passes_sasl_plain_credentials(self, *, app): + producer = Producer(app.transport) + assert app.in_transaction is False + with patch("aiokafka.AIOKafkaProducer") as AIOKafkaProducer: + AIOKafkaProducer.return_value.start = AsyncMock() + await producer.on_start() + AIOKafkaProducer.assert_called_once_with( + bootstrap_servers=["localhost:9098"], + client_id=f"faust-{faust.__version__}", + acks=-1, + linger_ms=0, + max_batch_size=16384, + max_request_size=1000000, + compression_type=None, + security_protocol="SASL_PLAINTEXT", + partitioner=producer.partitioner, + request_timeout_ms=1200000, + metadata_max_age_ms=300000, + connections_max_idle_ms=540000, + sasl_mechanism="PLAIN", + sasl_plain_username="uname", + sasl_plain_password="pw", + ssl_context=None, + transactional_id=None, + **({"api_version": "auto"} if _AIOKAFKA_HAS_API_VERSION else {}), + ) + def test__producer_type(self, *, producer, app): assert producer._producer_type is aiokafka.AIOKafkaProducer From 5fb4fdce04990a4c59efa4e64c1c720acdf97fda Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 28 Sep 2026 16:09:53 +0000 Subject: [PATCH 3/5] Clean up SASL regression test validation Co-authored-by: wbarnha <25623043+wbarnha@users.noreply.github.com> --- .coverage.runnervmtr4k5.pid4524.Xm0ekHtx | 0 tests/unit/transport/drivers/test_aiokafka.py | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) delete mode 100644 .coverage.runnervmtr4k5.pid4524.Xm0ekHtx diff --git a/.coverage.runnervmtr4k5.pid4524.Xm0ekHtx b/.coverage.runnervmtr4k5.pid4524.Xm0ekHtx deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index 75b1ebf58..b573ad0f6 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -1586,7 +1586,7 @@ async def test__new_producer__default(self, *, app): broker="kafka://localhost:9098", broker_credentials=auth.SASLCredentials( username="uname", - **{"pass" "word": "pw"}, + password=("p" "w"), mechanism="PLAIN", ), ) From 4b12caaef9298bfb680eeab15428f35e8b9ef862 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Wed, 16 Sep 2026 13:43:45 -0400 Subject: [PATCH 4/5] Validate SASL/PLAIN credentials before aiokafka setup --- faust/transport/drivers/aiokafka.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py index 8a17c1c7b..544dbd483 100644 --- a/faust/transport/drivers/aiokafka.py +++ b/faust/transport/drivers/aiokafka.py @@ -1702,6 +1702,15 @@ def credentials_to_aiokafka_auth( "ssl_context": credentials.ssl_context, } elif isinstance(credentials, SASLCredentials): + if credentials.mechanism.value == "PLAIN" and ( + credentials.username is None or credentials.password is None + ): + raise ImproperlyConfigured( + "SASL/PLAIN requires both username and password in " + "broker_credentials=SASLCredentials(...). " + "Check the credentials configured on the App that owns " + "the topic before creating its producer or consumer." + ) return { "security_protocol": credentials.protocol.value, "sasl_mechanism": credentials.mechanism.value, From f2d036de8a311163085404e4befd2ae55defbc1d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 28 Sep 2026 17:03:06 +0000 Subject: [PATCH 5/5] Test SASL/PLAIN validation rejects missing username or password Moved from #795. The forwarding case is already covered by the existing SASLCredentials(username="foo", password="bar") parametrization. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013m2okVri4b65RudZMXhYv2 --- tests/unit/transport/drivers/test_aiokafka.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index b573ad0f6..21df98711 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/tests/unit/transport/drivers/test_aiokafka.py @@ -2159,3 +2159,17 @@ def test_ensure_aiokafka_TPset(): actual = ensure_aiokafka_TPset({TP(topic="foo", partition=0)}) assert actual == {TopicPartition("foo", 0)} assert all(isinstance(tp, TopicPartition) for tp in actual) + + +@pytest.mark.parametrize( + "username,password", + [(None, None), (None, "bar"), ("foo", None)], +) +def test_credentials_to_aiokafka__sasl_plain_missing_credentials(username, password): + credentials = auth.SASLCredentials( + username=username, password=password, mechanism="PLAIN" + ) + with pytest.raises(ImproperlyConfigured, match="broker_credentials") as exc: + credentials_to_aiokafka_auth(credentials) + assert "foo" not in str(exc.value) + assert "bar" not in str(exc.value)