Как написать локальный kafka со способом авторизации юзеров: username, password, SASL_SSL, SCRAM-SHA-512?

Question

[Александр Сергеевич]

В моем compose.yaml прописаны сервисы для скрапинга (mysal, scrapy, rabbitmq), клиент хочет получать данные в kafka на своей стороне. Цель: расширить compose.yaml, дописав туда сервис kafka с соответствующими критериями авторизации, чтоб протестить отправку сообщений из своей БД в kafka.
На даном этапе я пытаюсь построить модель этого сервера.

kafka:
      image: bitnami/kafka:latest
      environment:
        - KAFKA_CFG_NODE_ID=0
        - KAFKA_CFG_PROCESS_ROLES=controller,broker
        - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
        - KAFKA_CFG_LISTENERS=SASL_SSL://:9092,CONTROLLER://:9093
        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
        - KAFKA_CFG_ADVERTISED_LISTENERS=SASL_SSL://localhost:9092
        - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
        - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=SASL_SSL
        - KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL=SCRAM-SHA-512
        - KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=SCRAM-SHA-512
        - KAFKA_CFG_SASL_SCRAM_USERS=user:my_pass
        - KAFKA_CFG_SASL_MECHANISM=SCRAM-SHA-512
        - KAFKA_CFG_SASL_ENABLED_MECHANISMS=SCRAM-SHA-512
        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL=SASL_SSL
        - KAFKA_TLS_TYPE=JKS
        - KAFKA_ENABLE_KRAFT=yes
        - KAFKA_CLIENT_USERS=user
        - KAFKA_CLIENT_PASSWORDS=my_pass
        - KAFKA_CONTROLLER_USER=user
        - KAFKA_CONTROLLER_PASSWORD=my_pass
        - KAFKA_INTER_BROKER_USER=user
        - KAFKA_INTER_BROKER_PASSWORD=my_pass
        - KAFKA_CERTIFICATE_PASSWORD=my_pass
        - KAFKA_CFG_SASL_JAAS_CONFIG='org.apache.kafka.common.security.scram.ScramLoginModule required username="user" password="my_pass" user_user="my_pass";'
      ports:
        - "9093:9093"
        - "9092:9092"
      volumes:
        - ./certs:/opt/bitnami/kafka/config/certs:ro

кафка разворачивается в контейнере но постоянно роняет:
INFO [SocketServer listenerType=BROKER, nodeId=0] Failed authentication with /172.23.0.1 (channelId=172.23.0.2:9092-172.23.0.1:58478-2-10) (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512) (org.apache.kafka.common.network.Selector)
ключи и самописный сертификат присутствуют:
├── certs
│   ├── ca.crt
│   ├── ca.key
│   ├── ca.pem
│   ├── kafka.crt
│   ├── kafka.key
│   ├── kafka.keystore.jks
│   ├── kafka.p12
│   └── kafka.truststore.jks
├── compose.yaml
├── convert_jks_to_pem.sh
├── gen_certs.sh
├── kafka_consumer.py
├── kafka_producer.py
└── ssl_config.py
Может у кого то есть параметры для развертывания данного детища с вышеописаными условиями авторизации? Или в моем детище что-то поправить нужно?