Help: NoBrokersAvailable

[wawava]

hi
in kafka_2.11-0.10.1.0 version use kafka-python to connect error

use bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning is correct
my python code:
from kafka import KafkaConsumer
consumer = KafkaConsumer('test',bootstrap_servers=['localhost:9092'])
for message in consumer:
print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value))
error message:
Traceback (most recent call last):
File "pkafka.py", line 5, in
bootstrap_servers=['localhost:9092'])
File "/usr/lib/python2.7/site-packages/kafka_python-1.3.2.dev0-py2.7.egg/kafka/consumer/group.py", line 287, in init
self._client = KafkaClient(metrics=self._metrics, **self.config)
File "/usr/lib/python2.7/site-packages/kafka_python-1.3.2.dev0-py2.7.egg/kafka/client_async.py", line 204, in init
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "/usr/lib/python2.7/site-packages/kafka_python-1.3.2.dev0-py2.7.egg/kafka/client_async.py", line 813, in check_version
raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

[wawava]

ss -tlun command find the port is open
tcp LISTEN 0 50 :::9092 :::*

[stevehouel]

Any news fro this issue. I also have it on my Centos with same configuration

[jayalane]

This probably means the Kafka cluster is not healthy; NoBrokersAvailable is a normal (tho errorish) condition. If not you need to provide a way to reproduce.

[Miando]

I have the same error with this version

[jeffwidman]

For those reporting this error, can you confirm that the console consumer shipped with kafka package is working when run on the same machine where kafka-python is failing?

This is needed to confirm that the issue is an actual kafka-python bug and not due to network misconfiguration or an unhealthy Kafka cluster, both of which would be totally valid reasons for this error.

[typehorror]

I'm on Ubuntu and facing the same issue. kafka and producer are on the same machine. I can produce and consume using the binaries from kafka

zookeeper:

[15:32] ┬ brice@brice-laptop ~/Downloads/kafka_2.11-0.10.2.0 
        └ $ bin/zookeeper-server-start.sh config/zookeeper.properties

Kafka

[15:33] ┬ brice@brice-laptop ~/Downloads/kafka_2.11-0.10.2.0 
        └ $ bin/kafka-server-start.sh config/server.properties

[15:34] ┬ brice@brice-laptop ~/Code/kafka 
        └ $ ipython 
WARNING: Attempting to work in a virtualenv. If you encounter problems, please install IPython inside the virtualenv.
Python 2.7.12 (default, Nov 19 2016, 06:48:10) 
Type "copyright", "credits" or "license" for more information.

IPython 4.0.0 -- An enhanced Interactive Python.
?         -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help      -> Python's own help system.
object?   -> Details about 'object', use 'object??' for extra details.

In [1]: from kafka import KafkaProducer

In [2]: producer = KafkaProducer(bootstrap_servers='localhost:9092')
---------------------------------------------------------------------------
NoBrokersAvailable                        Traceback (most recent call last)
<ipython-input-2-8261074fd9a6> in <module>()
----> 1 producer = KafkaProducer(bootstrap_servers='localhost:9092')

/home/brice/Code/kafka/venv/lib/python2.7/site-packages/kafka/producer/kafka.pyc in __init__(self, **configs)
    332 
    333         client = KafkaClient(metrics=self._metrics, metric_group_prefix='producer',
--> 334                              **self.config)
    335 
    336         # Get auto-discovered version from client if necessary

/home/brice/Code/kafka/venv/lib/python2.7/site-packages/kafka/client_async.pyc in __init__(self, **configs)
    202         if self.config['api_version'] is None:
    203             check_timeout = self.config['api_version_auto_timeout_ms'] / 1000
--> 204             self.config['api_version'] = self.check_version(timeout=check_timeout)
    205 
    206     def _bootstrap(self, hosts):

/home/brice/Code/kafka/venv/lib/python2.7/site-packages/kafka/client_async.pyc in check_version(self, node_id, timeout, strict)
    793             try_node = node_id or self.least_loaded_node()
    794             if try_node is None:
--> 795                 raise Errors.NoBrokersAvailable()
    796             self._maybe_connect(try_node)
    797             conn = self._conns[try_node]

NoBrokersAvailable: NoBrokersAvailable

In [3]: 

[Miando]

My issue was with connect Kafka to Zookeeper

[dpkp]

Hi all -- NoBrokersAvailable is raised in this scenario if bootstrapping metadata failed. Bootstrapping relies exclusively on bootstrap_servers . If you are seeing this error it is because kafka-python was unable to connect to any of the brokers you configured as bootstrap_servers.

If you want to find out more about what is happening under the covers, enable debug logging:

import logging
logging.basicConfig(level=logging.DEBUG)

In addition there will be some additional connection logging in the next release and that should make it even easier to diagnose networking problems like this.

[typehorror]

Thanks for the insight @dpkp , this is going to help me understand what is going on.

Just FYI I'm adding the result here

[16:24] ┬ brice@brice-laptop ~/Code/kafka 
        └ $ ipython
Python 2.7.12 (default, Nov 19 2016, 06:48:10) 
Type "copyright", "credits" or "license" for more information.

IPython 5.3.0 -- An enhanced Interactive Python.
?         -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help      -> Python's own help system.
object?   -> Details about 'object', use 'object??' for extra details.

In [1]: from kafka import KafkaConsumer, KafkaProducer

In [2]: import logging
   ...: logging.basicConfig(level=logging.DEBUG)
   ...: 

In [3]:     producer = KafkaProducer(bootstrap_servers='localhost:9092',
   ...:                              retries=5,
   ...:                              max_block_ms=10000,
   ...:                              compression_type='lz4',
   ...:                              value_serializer=str.encode)
DEBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
DEBUG:kafka.client:Attempting to bootstrap via node at localhost:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.conn:<BrokerConnection host=localhost/localhost port=9092>: creating new socket
DEBUG:kafka.conn:<BrokerConnection host=localhost/127.0.0.1 port=9092>: established TCP connection
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.conn:<BrokerConnection host=localhost/127.0.0.1 port=9092> Request 1: MetadataRequest_v0(topics=[])
DEBUG:kafka.conn:<BrokerConnection host=localhost/127.0.0.1 port=9092> Response 1: MetadataResponse_v0(brokers=[(node_id=2, host=u'brice-laptop', port=9094), (node_id=1, host=u'brice-laptop', port=9093), (node_id=0, host=u'brice-laptop', port=9092)], topics=[(error_code=0, topic=u'__consumer_offsets', partitions=[(error_code=0, partition=23, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=41, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=32, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=17, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=8, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=35, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=44, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=26, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=11, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=47, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=38, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=29, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=20, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=2, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=5, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=14, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=46, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=40, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=49, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=13, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=4, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=31, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=22, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=16, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=7, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=43, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=25, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=34, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=10, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=37, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=1, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=19, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=28, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=45, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=27, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=36, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=18, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=9, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=48, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=21, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=3, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=12, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=30, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=39, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=15, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=42, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=24, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=6, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=33, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=0, leader=0, replicas=[0], isr=[0])]), (error_code=0, topic=u'my-replicated-topic', partitions=[(error_code=0, partition=0, leader=0, replicas=[0, 1, 2], isr=[0, 1, 2])]), (error_code=0, topic=u'test', partitions=[(error_code=0, partition=0, leader=0, replicas=[0], isr=[0])])])
DEBUG:kafka.cluster:Updated cluster metadata to ClusterMetadata(brokers: 3, topics: 3, groups: 0)
DEBUG:kafka.client:Initiating connection to node 2 at brice-laptop:9094
DEBUG:kafka.metrics.metrics:Added sensor with name node-2.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-2.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-2.latency
DEBUG:kafka.conn:<BrokerConnection host=brice-laptop/brice-laptop port=9094>: creating new socket
ERROR:kafka.conn:Connect attempt to <BrokerConnection host=brice-laptop/fe80::82fa:5bff:fe0a:9fc7%eth0 port=9094> returned error 22. Disconnecting.
WARNING:kafka.client:Node 2 connection failed -- refreshing metadata
DEBUG:kafka.client:Initiating connection to node 0 at brice-laptop:9092
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.latency
WARNING:kafka.client:Node 0 connection failed -- refreshing metadata
DEBUG:kafka.client:Initiating connection to node 1 at brice-laptop:9093
DEBUG:kafka.metrics.metrics:Added sensor with name node-1.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-1.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-1.latency
WARNING:kafka.client:Node 1 connection failed -- refreshing metadata
ERROR:kafka.client:No nodes found in metadata -- retrying bootstrap
DEBUG:kafka.client:Sleeping 0.0354 before bootstrapping again
DEBUG:kafka.client:Attempting to bootstrap via node at localhost:9092
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.cluster:Updated cluster metadata to ClusterMetadata(brokers: 3, topics: 3, groups: 0)
---------------------------------------------------------------------------
NoBrokersAvailable                        Traceback (most recent call last)
<ipython-input-3-1417523cc14e> in <module>()
      3                          max_block_ms=10000,
      4                          compression_type='lz4',
----> 5                          value_serializer=str.encode)

/home/brice/Code/kafka/venv/local/lib/python2.7/site-packages/kafka/producer/kafka.pyc in __init__(self, **configs)
    332 
    333         client = KafkaClient(metrics=self._metrics, metric_group_prefix='producer',
--> 334                              **self.config)
    335 
    336         # Get auto-discovered version from client if necessary

/home/brice/Code/kafka/venv/local/lib/python2.7/site-packages/kafka/client_async.pyc in __init__(self, **configs)
    202         if self.config['api_version'] is None:
    203             check_timeout = self.config['api_version_auto_timeout_ms'] / 1000
--> 204             self.config['api_version'] = self.check_version(timeout=check_timeout)
    205 
    206     def _bootstrap(self, hosts):

/home/brice/Code/kafka/venv/local/lib/python2.7/site-packages/kafka/client_async.pyc in check_version(self, node_id, timeout, strict)
    793             try_node = node_id or self.least_loaded_node()
    794             if try_node is None:
--> 795                 raise Errors.NoBrokersAvailable()
    796             self._maybe_connect(try_node)
    797             conn = self._conns[try_node]

NoBrokersAvailable: NoBrokersAvailable

[typehorror]

In case someone faces the exact same issue as I did, I believe my kafka config was the issue. I've added an explicit listener line to my config config/server.properties and it resolved the issue.

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:9092

[dpkp]

Actually the logs suggest that in fact there is a kafka-python error handling your original IPv6 configuration. This line in particular stands out:

ERROR:kafka.conn:Connect attempt to <BrokerConnection host=brice-laptop/fe80::82fa:5bff:fe0a:9fc7%eth0 port=9094> returned error 22. Disconnecting.

Socket Error 22 is and "Invalid Argument" error, which suggests that we are not able to connect to ipv6 addresses that have a scope-id (%eth0). I will open a separate issue for this.

[typehorror]

If it helps, I was following the Kafka quickstart tutorial (https://kafka.apache.org/quickstart#quickstart_multibroker) where you spawn 3 Kafka servers locally on port 9092, 9093 and 9094. During the tutorial, you have to set the listeners variable on servers 2 and 3 but nothing is said regarding the initial server (9092).