kafka-python поднять ошибку UnrecognizedBrokerVersion

Я получаю эту ошибку при создании KafkaProducer с kafka-python пакет:

[ERROR] UnrecognizedBrokerVersion: UnrecognizedBrokerVersion
Traceback (most recent call last):
  File "/var/lang/lib/python3.7/imp.py", line 234, in load_module
    return load_source(name, filename, file)
  File "/var/lang/lib/python3.7/imp.py", line 171, in load_source
    module = _load(spec)
  File "<frozen importlib._bootstrap>", line 696, in _load
  File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/var/task/kafka/producer/kafka.py", line 381, in __init__
    **self.config)
  File "/var/task/kafka/client_async.py", line 240, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/var/task/kafka/client_async.py", line 908, in check_version
    version = conn.check_version(timeout=remaining, strict=strict, topics=list(self.config['bootstrap_topics_filter']))
  File "/var/task/kafka/conn.py", line 1228, in check_version
    raise Errors.UnrecognizedBrokerVersion()

Код выглядит следующим образом:

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=os.environ.get('KAFKA_HOST', 'localhost:9092'))

Я использую Python 3.7 и кластер AWS MSK.

Источник

2 ответа

Решение

Решил, просто добавив security_protocol="SSL" к KafkaProducer следующим образом:

from kafka import KafkaProducer
producer = KafkaProducer(security_protocol="SSL", bootstrap_servers=os.environ.get('KAFKA_HOST', 'localhost:9092'))
Источник

Есть много дополнительных параметров с kafka-pythonlib методы. Вот недавний рабочий скрипт, с помощью которого я смог прослушать сервер Azure:

      import kafka  # pip install kafka-python

consumer = kafka.KafkaConsumer("test-topic",
    bootstrap_servers=["test-server.servicebus.windows.net:9093"],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='$Default',
    sasl_mechanism='PLAIN',
    sasl_plain_password=os.environ["KAFKA_PASSWORD"],
    sasl_plain_username='$ConnectionString',
    security_protocol='SASL_SSL',
    value_deserializer=lambda x: x.decode('utf-8'))

print(datetime.datetime.now())
for message in consumer:
    message_dict = json.loads(message.value)
    print(datetime.datetime.now())
    print("%s" % (json.dumps(message_dict,indent=4)))

Это грубо, но эффективно. Проще, чем демонстрационный потребитель Kafka, обернутый в Java, со страницы https://kafka.apache.org/quickstart, для которого требовались различные файлы конфигурации и переменные среды.

Источник
Другие вопросы по тегам