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,
diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py
index 0f7173f4f..21df98711 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",
+            password=("p" "w"),
+            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
 
@@ -2122,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)