Kafka-蟒蛇：生产者无法连接

Question

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

Python应用程序过去也运行良好，但是在动物园管理员重新启动后，它不再能够连接。

我使用文档中的裸露骨骼示例：

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

在我的例子中，_metadata_refresh_in_progress是False，但是ttl（）=0；

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

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

有什么建议吗？

Answer 1

我也有同样的问题。

我用user3503929的提示解决了问题。

kafka服务器安装在windows上。

服务器属性

...
host.name = 0.0.0.0
...

.

producer = KafkaProducer(bootstrap_servers='192.168.1.3:9092',         
                                         value_serializer=str.encode)
producer.send('test', value='aaa')
producer.close()
print("DONE.")

windows kafka客户端的处理没有问题。然而，当我在ubuntu中使用kafka-python向topic发送消息时，出现了< code>NoBrokersAvailable异常。

将以下设置添加到服务器属性。

...
advertised.host.name = 192.168.1.3
...

它在相同的代码中成功运行。为此我花了三个小时。

谢谢

Answer 2

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

Answer 3

我也有同样的问题，上面的解决方案都不管用。然后我阅读异常消息，似乎必须指定api_version，所以

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

注意：与 kafka 版本 1.0.0 匹配的元组 （1，0，0）

工作正常（至少完成无例外，现在必须说服它接受消息；））