Python 経由でメッセージを kafka トピックに公開しようとしていますが、エラーが発生します。CLI 経由で接続して公開できます。いくつかのガイダンスを期待しています。私はグーグルでドキュメントを読みました。ありがとう!!
成功した CLI コマンド:
kafka-console-producer --broker-list
123.45.67.891:1234,123.45.67.892:1234,123.45.67.893:1234 --
producer.config C:\Users\fake_user\Kafka\client-ssl.properties --topic FakeTopic
Contents of client-ssl.properties:
security.protocol = SSL
ssl.truststore.location = C:/Users/fake_user/Kafka/kafka-truststore
ssl.truststore.password = fakepass
コード:
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['123.45.67.891:1234', '123.45.67.892:1234', '123.45.67.893:1234'],
security_protocol='SSL',
ssl_certfile=r'C:\Users\fake_user\Kafka\kafka-truststore',
ssl_password='fakepass')
producer.send('FakeTopic', value='python_test', key='test')
結果エラー:
Traceback (most recent call last):
File "kafka_post_test.py", line 6, in <module>
ssl_password='fakepass')
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\producer\kafka.py", line 381, in __init__
**self.config)
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\client_async.py", line 239, in __init__
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\client_async.py", line 874, in check_version
version = conn.check_version(timeout=remaining, strict=strict, topics=list(self.config['bootstrap_topics_filter']))
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\conn.py", line 1078, in check_version
if not self.connect_blocking(timeout_at - time.time()):
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\conn.py", line 331, in connect_blocking
self.connect()
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\conn.py", line 420, in connect
if self._try_handshake():
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\site-packages\kafka\conn.py", line 496, in _try_handshake
self._sock.do_handshake()
File "C:\Users\fake_user\AppData\Local\Programs\Python\Python37-32\lib\ssl.py", line 1117, in do_handshake
self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate in certificate chain (_ssl.c:1051)