[BUG] SASL PLAINTEXT producer initialization fails : missing sasl_plain_username/password #794

app = faust.App(id="super-app",
                           broker="kafka://localhost:9098,
                           store="rocksdb://",
                           processing_guarantee="exactly_once",
                           broker_credentials=faust.SASLCredentials(
                               username="some_username",
                               password="some_password",
                               mechanism="PLAIN"
                           ))

output_topic = app.topic("my_output_topic")
await topic.send(value="some_data")

..\.venv\Lib\site-packages\faust\topics.py:190: in send
    return await self._send_now(
..\.venv\Lib\site-packages\faust\channels.py:315: in _send_now
    return await self.publish_message(
..\.venv\Lib\site-packages\faust\topics.py:411: in publish_message
    producer = await self._get_producer()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
..\.venv\Lib\site-packages\faust\topics.py:388: in _get_producer
    return await self.app.maybe_start_producer()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
..\.venv\Lib\site-packages\mode\utils\futures.py:48: in __call__
    result = await self.fun(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
..\.venv\Lib\site-packages\faust\app\base.py:1694: in maybe_start_producer
    await producer.maybe_start()
..\.venv\Lib\site-packages\mode\services.py:894: in maybe_start
    await self.start()
..\.venv\Lib\site-packages\mode\services.py:827: in start
    await self._default_start()
..\.venv\Lib\site-packages\mode\services.py:834: in _default_start
    await self._actually_start()
..\.venv\Lib\site-packages\mode\services.py:857: in _actually_start
    await self.on_start()
..\.venv\Lib\site-packages\faust\transport\drivers\aiokafka.py:1386: in on_start
    producer = self._producer = self._new_producer()
                                ^^^^^^^^^^^^^^^^^^^^
..\.venv\Lib\site-packages\faust\transport\drivers\aiokafka.py:1331: in _new_producer
    return self._producer_type(
..\.venv\Lib\site-packages\aiokafka\producer\producer.py:296: in __init__
    self.client = AIOKafkaClient(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <[AttributeError("'AIOKafkaClient' object has no attribute '_client_id'") raised in repr()] AIOKafkaClient object at 0x1d47657fe00>
loop = <ProactorEventLoop running=False closed=False debug=False>
bootstrap_servers = ['127.0.0.1:9092'], client_id = 'faust-0.15.3'
metadata_max_age_ms = 300000, request_timeout_ms = 1200000
retry_backoff_ms = 100, ssl_context = None, security_protocol = 'SASL_PLAINTEXT'
api_version = None, connections_max_idle_ms = 540000, sasl_mechanism = 'PLAIN'
sasl_plain_username = None, sasl_plain_password = None
sasl_kerberos_service_name = 'kafka', sasl_kerberos_domain_name = None
sasl_oauth_token_provider = None

    def __init__(
        self,
        *,
        loop=None,
        bootstrap_servers="localhost",
        client_id="aiokafka-" + __version__,
        metadata_max_age_ms=300000,
        request_timeout_ms=40000,
        retry_backoff_ms=100,
        ssl_context=None,
        security_protocol="PLAINTEXT",
        api_version=None,
        connections_max_idle_ms=540000,
        sasl_mechanism="PLAIN",
        sasl_plain_username=None,
        sasl_plain_password=None,
        sasl_kerberos_service_name="kafka",
        sasl_kerberos_domain_name=None,
        sasl_oauth_token_provider=None,
    ):
        if loop is None:
            loop = get_running_loop()
    
        if api_version is not None:
            warnings.warn(
                "The `api_version` parameter has been deprecated since 0.13.0. "
                "It is now a no-op and will be removed in a future release. ",
                DeprecationWarning,
                stacklevel=2,
            )
    
        if security_protocol not in ("SSL", "PLAINTEXT", "SASL_PLAINTEXT", "SASL_SSL"):
            raise ValueError("`security_protocol` should be SSL or PLAINTEXT")
        if security_protocol in ["SSL", "SASL_SSL"] and ssl_context is None:
            raise ValueError("`ssl_context` is mandatory if security_protocol=='SSL'")
        if security_protocol in ["SASL_SSL", "SASL_PLAINTEXT"]:
            if sasl_mechanism not in (
                "PLAIN",
                "GSSAPI",
                "SCRAM-SHA-256",
                "SCRAM-SHA-512",
                "OAUTHBEARER",
            ):
                raise ValueError(
                    "only `PLAIN`, `GSSAPI`, `SCRAM-SHA-256`, "
                    "`SCRAM-SHA-512` and `OAUTHBEARER`"
                    "sasl_mechanism are supported "
                    "at the moment"
                )
            if sasl_mechanism == "PLAIN" and (
                sasl_plain_username is None or sasl_plain_password is None
            ):
>               raise ValueError(
                    "sasl_plain_username and sasl_plain_password required for "
                    "PLAIN sasl"
                )
E               ValueError: sasl_plain_username and sasl_plain_password required for PLAIN sasl

..\.venv\Lib\site-packages\aiokafka\client.py:132: ValueError