我正在尝试连接到上的bluemix消息中心示例http://bluemix.net. 这个简单的脚本
# !/usr/bin/env python
from kafka import KafkaProducer
from kafka.errors import KafkaError
kafka_brokers_sasl = [
"kafka01-prod01.messagehub.services.us-south.bluemix.net:9093",
"kafka02-prod01.messagehub.services.us-south.bluemix.net:9093",
"kafka03-prod01.messagehub.services.us-south.bluemix.net:9093",
"kafka04-prod01.messagehub.services.us-south.bluemix.net:9093",
"kafka05-prod01.messagehub.services.us-south.bluemix.net:9093" ]
sasl_plain_username = "xxxxxxxxxxxxxxx"
sasl_plain_password = "xxxxxxxxxxxxxxxxxxxxxxxxx"
sasl_mechanism = 'SASL_PLAINTEXT'
producer = KafkaProducer(bootstrap_servers = kafka_brokers_sasl,
sasl_plain_username = sasl_plain_username,
sasl_plain_password = sasl_plain_password,
sasl_mechanism = sasl_mechanism )
以以下例外结束:
Traceback (most recent call last):
File "./test-mh.py", line 12, in <module>
producer = KafkaProducer(bootstrap_servers = kafka_brokers_sasl, sasl_plain_username = sasl_plain_username, sasl_plain_password = sasl_plain_password, sasl_mechanism = sasl_mechanism )
File "/usr/local/lib/python2.7/dist-packages/kafka/producer/kafka.py", line 328, in __init__
**self.config)
File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 202, in __init__
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 791, in check_version
raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable
我从messagehub服务凭证对象中获得了kafka\u brokers\u sasl、sasl\u plain\u username和sasl\u plain\u password。我使用的是kafkapython1.3.1,它似乎支持sasl身份验证机制。你知道我做错了什么吗?谢谢。
1条答案
按热度按时间kfgdxczn1#
MessageHub要求客户端使用TLS1.2连接进行连接。这意味着指定
security_protocol
参数到KafkaProducer
同时也是一个ssl.SSLContext
通过ssl_context
参数-看起来python kafka客户机创建了一个SSLv23
默认情况下为上下文。以下是连接所需的更改: