I am getting this error when constructing KafkaProducer with the kafka-python
package:
JavaScript
x
19
19
1
[ERROR] UnrecognizedBrokerVersion: UnrecognizedBrokerVersion
2
Traceback (most recent call last):
3
File "/var/lang/lib/python3.7/imp.py", line 234, in load_module
4
return load_source(name, filename, file)
5
File "/var/lang/lib/python3.7/imp.py", line 171, in load_source
6
module = _load(spec)
7
File "<frozen importlib._bootstrap>", line 696, in _load
8
File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
9
File "<frozen importlib._bootstrap_external>", line 728, in exec_module
10
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
11
File "/var/task/kafka/producer/kafka.py", line 381, in __init__
12
**self.config)
13
File "/var/task/kafka/client_async.py", line 240, in __init__
14
self.config['api_version'] = self.check_version(timeout=check_timeout)
15
File "/var/task/kafka/client_async.py", line 908, in check_version
16
version = conn.check_version(timeout=remaining, strict=strict, topics=list(self.config['bootstrap_topics_filter']))
17
File "/var/task/kafka/conn.py", line 1228, in check_version
18
raise Errors.UnrecognizedBrokerVersion()
19
The code is as follows:
JavaScript
1
3
1
from kafka import KafkaProducer
2
producer = KafkaProducer(bootstrap_servers=os.environ.get('KAFKA_HOST', 'localhost:9092'))
3
I am using Python 3.7 and an AWS MSK cluster.
Advertisement
Answer
Solved it by just adding security_protocol="SSL"
to the KafkaProducer
as follows:
JavaScript
1
3
1
from kafka import KafkaProducer
2
producer = KafkaProducer(security_protocol="SSL", bootstrap_servers=os.environ.get('KAFKA_HOST', 'localhost:9092'))
3