kafka-python:生产者无法连接

Question

kafka-python (1.0.0)在连接到代理时抛出错误。同时，/usr/bin/kafka-console-producer和/usr/bin/kafka-console-consumer可以正常工作。

Python应用程序过去也工作得很好，但在zookeeper重启后，它不再可以连接。

我使用的是文档中的基本示例：

from kafka import KafkaProducer
from kafka.common import KafkaError

producer = KafkaProducer(bootstrap_servers=['hostname:9092'])

# Asynchronous by default
future = producer.send('test-topic', b'raw_bytes')

我得到了这个错误：

Traceback (most recent call last):   File "pp.py", line 4, in <module>
    producer = KafkaProducer(bootstrap_servers=['hostname:9092'])   File "/usr/lib/python2.6/site-packages/kafka/producer/kafka.py", line 246, in __init__
    self.config['api_version'] = client.check_version()   File "/usr/lib/python2.6/site-packages/kafka/client_async.py", line 629, in check_version
    connect(node_id)   File "/usr/lib/python2.6/site-packages/kafka/client_async.py", line 592, in connect
    raise Errors.NodeNotReadyError(node_id) kafka.common.NodeNotReadyError: 0 Exception AttributeError: "'KafkaProducer' object has no attribute '_closed'" in <bound method KafkaProducer.__del__ of <kafka.producer.kafka.KafkaProducer object at 0x7f6171294c50>> ignored

单步执行( /usr/lib/python2.6/site-packages/kafka/client_async.py)时，我注意到第270行的计算结果为false：

270         if not self._metadata_refresh_in_progress and not self.cluster.ttl() == 0:
271             if self._can_send_request(node_id):
272                 return True
273         return False

在我的例子中，self._metadata_refresh_in_progress是假的，但是ttl() = 0；

与此同时，kafka-console-*正在愉快地推送消息：

/usr/bin/kafka-console-producer --broker-list hostname:9092 --topic test-topic
hello again
hello2

有什么建议吗？

Answer 1

我遇到了同样的问题，上面的解决方案都不起作用。然后，我阅读了异常消息，似乎必须指定api_version，因此

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],api_version=(0,1,0))

说明:与kafka版本1.0.0匹配的元组(1,0,0)

工作正常(至少完成时没有异常，现在必须说服它接受消息;)

Answer 2

我也遇到过类似的问题。在我的例子中，代理主机名在客户端无法解析。尝试在配置文件中显式设置advertised.host.name。

Answer 3

一台主机可以有多个dns别名。它们中的任何一个都可以用于ssh或ping测试。但是，kafka连接应该使用与broker的server.properties文件中的advertised.host.name匹配的别名。

我在bootstrap_servers参数中使用了不同的别名。因此出现了错误。一旦我将调用更改为使用advertised.hostname，问题就解决了