Apache Beam с Python. Ошибка при попытке аутентификации Apache Kafka SASL_SSL OAUTHBEARER ⇐ Python

[Anonymous]

Я разрабатываю конвейер в Apache Beam 2.45.0 с использованием Python 3.8.
Мне нужно прочитать данные из Apache Kafka, используя метод ReadfromKafka.
Сервер Apache Kafka защищен протоколом SASL_SSL с использованием механизма OAUTHBEARER.
Я разработал CallBackHandler с целью управления токенами с помощью Authn Сервер (он работает на простом потребителе kafka с библиотекой kafka-python), но когда я пробую это на Apache Beam через поток данных, я получаю следующую ошибку:

Код: Выделить всё

Caused by: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set

Это мой код конвейера:

Код: Выделить всё

p = beam.Pipeline(options=pipeline_options)

consumer_config = {'bootstrap.servers': self.bootstrap_servers,
'auto.offset.reset': 'earliest',
'security.protocol': 'SASL_SSL',
'sasl.mechanism': 'OAUTHBEARER',
'sasl.oauth.token.provider': TokenProvider(self.project_id, self.secret_id) # CallBackHandler class
}

messages = (
p
| "Read from Kafka" >> beam.io.kafka.ReadFromKafka(consumer_config=consumer_config, topics=[self.topic])
)

Есть идеи?
Спасибо!

Подробнее здесь: https://stackoverflow.com/questions/756 ... l-ssl-oaut