Kafka-python: производитель не может подключиться

Ответ 1

У меня была такая же проблема, и ни одно из вышеупомянутых решений не сработало. Затем я читаю сообщения об исключениях, и кажется, что обязательно указывать api_version, поэтому

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

примечание: кортеж (1,0,0) соответствует версии kafka 1.0.0

работает нормально (по крайней мере, завершается без исключений, теперь нужно убедить его принять сообщения;))

Ответ 2

У меня была похожая проблема. В моем случае имя хоста брокера было неразрешимым на стороне клиента. Попробуйте явно указать advertised.host.name в файле конфигурации.

Ответ 3

Хост может иметь несколько псевдонимов DNS. Любой из них будет работать для теста ssh или ping. Однако соединение kafka должно использовать псевдоним, который соответствует advertised.host.name в файле server.properties брокера.

Я использовал другой псевдоним в параметре bootstrap_servers. Отсюда и ошибка. После того, как я изменил вызов на использование advertised.hostname, проблема была решена

Ответ 4

У меня была та же проблема.

Я решил проблему с подсказкой user3503929.

Сервер kafka был установлен на windows.

server.properties

...
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.")

Не было проблем с обработкой в ​​клиенте kafka. Однако, когда я отправляю сообщение в тему с помощью kafka-python в ubuntu, возникает исключение NoBrokersAvailable.

Добавьте следующие параметры в server.properties.

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

Он успешно работает в том же коде. Из-за этого я провел три часа.

Спасибо

Ответ 5

У меня была аналогичная проблема, и мне помог удалить порт из bootstrap_servers.

consumer = KafkaConsumer('my_topic',
                     #group_id='x',
                     bootstrap_servers='kafka.com')

Ответ 6

Установите kafka-python с помощью pip install kafka-python

Шаги по созданию конвейера данных кафки: -
1. Запустите Zookeeper с помощью команды оболочки или установите zookeeperd с помощью

sudo apt-get install zookeeperd 

Это приведет к запуску zookeeper в качестве демона и по умолчанию прослушивает порт 2181

2. Запустите сервер kafka
3. Запустите script с помощью users.py и consumer.py на отдельных консолях, чтобы просмотреть данные в реальном времени.

Вот команды для запуска: -

cd kafka-directory
./bin/zookeeper-server-start.sh  ./config/zookeeper.properties    
./bin/kafka-server-start.sh  ./config/server.properties

Теперь, когда у вас запущен сервер zookeeper и kafka, запустите файл manufacturer.py script и consumer.py

Producer.py:

from kafka import KafkaProducer
import time

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
topic = 'simple-text'
<implement a file reading functionality or any log reader and put it in lines>
for line in lines:
lst = line.split(" ")
try:
    final_list = [lst[x] for x in range(14)]
    producer.send(topic, final_list[0]).get(timeout=10)
except IndexError as e:
    print e
    continue

Consumer.py: -

from kafka import KafkaConsumer
topic = 'simple-text'
consumer = KafkaConsumer(topic, bootstrap_servers=['localhost:9092'])
for message in consumer:
    # message value and key are raw bytes -- decode if necessary!
    # e.g., for unicode: `message.value.decode('utf-8')`
    # print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
    #                                       message.offset, message.key,
    #                                       message.value))
    print message

Теперь запустите файл manufacturer.py и consumer.py в отдельных терминалах, чтобы увидеть текущие данные..!

Примечание. Выше product.py script запускается один раз, чтобы запускать его навсегда, используйте цикл while и time time.

Ответ 7

В файле server.properties убедитесь, что для его IP-адреса получателя установлен IP-адрес вашего окна, доступный для удаленной машины. По умолчанию это localhost

Обновите эту строку на сервере server.properties:

listeners=PLAINTEXT://<Your-IP-address>:9092

Также убедитесь, что у вас нет брандмауэра, который может блокировать другие IP-адреса для доступа к вам. Если у вас есть sudo превалирует. Попробуйте отключить брандмауэр.

sudo systemctl stop firewalld

Ответ 8

Я знаю эту проблему и решение. Если вы протестировали ping и ssh, вы можете попытаться подключиться к серверу с помощью telnet.

Я думаю, что брандмауэр блокирует ваши сообщения. Попробуйте отключить межсетевой экран. Если ваши сообщения достигнут, отредактируйте разрешения брандмауэра.

sudo ufw disable

Но это решение очень зеленое.