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/test_aiokafka_sasl.py b/tests/unit/test_aiokafka_sasl.py
new file mode 100644
index 000000000..ea85bb828
--- /dev/null
+++ b/tests/unit/test_aiokafka_sasl.py
@@ -0,0 +1,43 @@
+import pytest
+
+from faust import SASLCredentials
+from faust.exceptions import ImproperlyConfigured
+from faust.transport.drivers.aiokafka import credentials_to_aiokafka_auth
+
+
+def test_plain_credentials_are_forwarded_to_aiokafka() -> None:
+    settings = credentials_to_aiokafka_auth(
+        SASLCredentials(
+            username="test-user",
+            password="test-password",
+            mechanism="PLAIN",
+        )
+    )
+
+    assert settings == {
+        "security_protocol": "SASL_PLAINTEXT",
+        "sasl_mechanism": "PLAIN",
+        "sasl_plain_username": "test-user",
+        "sasl_plain_password": "test-password",
+        "ssl_context": None,
+    }
+
+
+@pytest.mark.parametrize(
+    "username,password",
+    [(None, None), (None, "test-password"), ("test-user", None)],
+)
+def test_plain_credentials_require_username_and_password(
+    username: str | None, password: str | None
+) -> None:
+    credentials = SASLCredentials(
+        username=username,
+        password=password,
+        mechanism="PLAIN",
+    )
+
+    with pytest.raises(ImproperlyConfigured, match="broker_credentials") as exc:
+        credentials_to_aiokafka_auth(credentials)
+
+    assert "test-user" not in str(exc.value)
+    assert "test-password" not in str(exc.value)