How to specify user name and password for a FastKafka connection

[Jrqz]

Hi, I have inherited a partly/mostly written python Unix app which uses this FastKafka module. I was given a user name and password which the app's Producer presumably needs to send messages to the remote Kafka server. But nothing I can see in the FaskKafka code or documentation, nor AIOKafka which I gather FastKafka wraps, mentions user names and passwords!

I know little about Kafka, but I gather there is something called SASL which can configure this via a file. I would prefer to use environment variables if possible, and specify these items in the code like my other connect parameters, e.g.:

` def get_default_config(self):

    kafka_brokers: dict = {

        "local": {

            "description": "local development kafka broker",

            "url": os.environ.get("KAFKA_BROKER_URL", "localhost"),

            "port": os.environ.get("KAFKA_BROKER_PORT", 9092),

            "protocol": "kafka-secure",

            "security": {"type": "plain"},

            "username": os.environ.get("KAFKA_BROKER_USERNAME", None),   # ???!

            "password": os.environ.get("KAFKA_BROKER_PASSWORD", None),   # ???!

        },

    }

    return {

        "kafka_brokers": kafka_brokers,

        "title": "SF Ingestion",

    }`

(Although this config says "local broker", the URL is actually remote, in a development staging area.)

I am not fussy though. So I would very much appreciate it is someone could indicate the recommended way to specify these items. TIA

[kumaranvpl]

Hello @Jrqz,

I have noticed that your staging Kafka setup likely includes security measures such as usernames, passwords, and encryption. In light of this, the code to initialize the FastKafka object should resemble the following:

import os
from fastkafka import FastKafka
from aiokafka.helpers import create_ssl_context

kafka_brokers = {
    "local": {
        "description": "local development kafka broker",
        "url": os.environ.get("KAFKA_BROKER_URL", "localhost"),
        "port": os.environ.get("KAFKA_BROKER_PORT", 9092),
        "protocol": "kafka-secure",
        "security": {"type": "scramSha256"},
    },
}

fastkafka_app = FastKafka(
    title="Demo Kafka app",
    security_protocol=os.environ["KAFKA_SECURITY_PROTOCOL"],
    sasl_mechanism=os.environ["KAFKA_SASL_MECHANISM"],
    sasl_plain_username=os.environ["KAFKA_BROKER_USERNAME"],
    sasl_plain_password=os.environ["KAFKA_BROKER_PASSWORD"],
    ssl_context=create_ssl_context(),
    kafka_brokers=kafka_brokers,
)

In the provided code, the kafka_brokers dictionary is utilized for generating AsyncAPI specifications and documentation. The actual Kafka credentials should be specified during the initialization of the FastKafka object. Additionally, it's important to ensure accurate values are provided for the security_protocol and sasl_mechanism parameters, as these can differ based on the configuration of your Kafka instances.

I'd like to note that FastKafka is built on top of the AIOKafka project. If you wish to pass more arguments to AIOKafka, you can easily refer to AIOKafka's producer documentation and supply these arguments as keyword arguments during the initialization of the FastKafka object.

Please review the code and make any necessary adjustments based on your specific setup. If you have any further questions or concerns, feel free to ask.

[Jrqz]

Ah yes, sorry. My config is a bit misleading because the broker is remote in an Azure cloud.

Also while I think of it (although this is a different issue) there seems to be a bug in the FastKafka Tester functionality, in that it fails if the topic ID contains any hyphens (which are allowed by the Kafka standards).

For example, the output, for a topic name of "chatoptin-sf-connector" is as follows (and this error vanishes when the hyphens are replaced by underscores) :

  File "/usr/local/lib/python3.11/unittest/async_case.py", line 90, in _callTestMethod
    if self._callMaybeAsync(method) is not None:
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/unittest/async_case.py", line 112, in _callMaybeAsync
    return self._asyncioRunner.run(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/unittest/mock.py", line 1392, in patched
    return await func(*newargs, **newkeywargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/tests/test_ingestion.py", line 84, in test_main_date_range
    async with Tester(kafka_service.kafka_app):
  File "/usr/local/lib/python3.11/site-packages/fastkafka/_application/tester.py", line 118, in __aenter__
    return await self._ctx.__aenter__()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/contextlib.py", line 204, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/fastkafka/_application/tester.py", line 104, in _create_ctx
    await self._start_tester()
  File "/usr/local/lib/python3.11/site-packages/fastkafka/_application/tester.py", line 83, in _start_tester
    self.create_mocks()
  File "/usr/local/lib/python3.11/site-packages/fastkafka/_application/app.py", line 1031, in create_mocks
    self.AppMocks = namedtuple(  # type: ignore
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/collections/__init__.py", line 393, in namedtuple
    raise ValueError('Type names and field names must be valid '
ValueError: Type names and field names must be valid identifiers: 'mirror_139850039076816_on_chatoptin-sf-connector_5729812687901956768'```

[kumaranvpl]

Hello @Jrqz,

Internally we are using namedtuple in FastKafka Tester. Looks like this error is arising from namedtuple module. We will look into this issue.

Thanks for letting us know.

[Jrqz]

Many thanks Kumaran. It would be good if you (or someone) could also when convenient look into why that producer_store list in producer_decorator.py has its second element as None, when elsewhere the code attempts to use this value as a producer object. For reference, this is the config passed to the FastKafka() class:

        kafka_brokers: dict = {
            "local": {
                "description": "local development kafka broker",
                "url": config.KAFKA_BROKER_URL,
                "port": int(config.KAFKA_BROKER_PORT),
                "protocol": "kafka-secure",
                "security": {"type": "scramSha256"},
            },
        }
        return {
            "title": "SF Ingestion",
            "security_protocol": config.KAFKA_SECURITY_PROTOCOL,
            "sasl_mechanism": config.KAFKA_SASL_MECHANISM,
            "sasl_plain_username": config.KAFKA_BROKER_USERNAME,
            "sasl_plain_password": config.KAFKA_BROKER_PASSWORD,
            "ssl_context": create_ssl_context(),
            "kafka_brokers": kafka_brokers,
        }

I tried adding a "production" dict below the "local" dict in the kafka_broker list. But that made no difference

On the assumption that the four element producer_store list in producer_decorator.py was maybe (?) supposed to represent the slots for up to four Kafka brokers, I also tried slightly amending the producer_decorator.py code to use the first (and in my example only non-None) element of the producer_store list as the producer object. But that didn't work either. It was rather a silly assumption though, more a stab in the dark, because why would the code always choose only the second element in the list?!

(To repeat what I mentioned earlier, that name "local" for the Kafka broker is misleading because the URL is actually a remote broker!)

[sternakt]

Hi!
The failing of tester because of the hyphen in the topic name has been fixed in https://github.com/airtai/fastkafka/issues/472
You can expect it to be working in the next release :)

About the producers_store: the second element in the producers_list is a reference to the concrete producer, this is set to None when you create your application by design, when FastKafka app is started, it will call a populate_producers method which will then replace this None value with the Producer object.