Help: Problems connecting to broker

[alihossein]

hi
i using:
kafka-python :V 1.4.1
kafka broker : V 2.12-1.0.0
when run this code to connect to remote server:

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='x.x.251.60:9092')
producer.send('test-topic-222', b'some_message_bytes')
producer.flush()

I get this error :

Traceback (most recent call last):
  File "/var/www/html/kafka_python/producer.py", line 2, in <module>
    producer = KafkaProducer(bootstrap_servers='x.x.251.60:9092')
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/producer/kafka.py", line 363, in __init__
    **self.config)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 234, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 856, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

else I did not uncomment in #listeners=PLAINTEXT://:9092 in server.properties

[dpkp]

Can you post debug logs? Are you using SSL or SASL for your broker configuration?

[alihossein]

Are you using SSL or SASL for your broker configuration?

No

this is my log :

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
INFO:kafka.client:Bootstrapping cluster metadata from [('x.x.251.58', 9092, <AddressFamily.AF_INET: 2>)]
DEBUG:kafka.client:Attempting to bootstrap via node at x.x.251.58: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 node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: creating new socket
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: setting socket option (6, 1, 1)
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: connecting to x.x.251.58:9092
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: established TCP connection
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: Connection complete.
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.protocol.parser:Sending request MetadataRequest_v0(topics=[])
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092> Request 1: MetadataRequest_v0(topics=[])
DEBUG:kafka.protocol.parser:Received correlation id: 1
DEBUG:kafka.protocol.parser:Processing response MetadataResponse_v0
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092> Response 1 (154.2062759399414 ms): MetadataResponse_v0(brokers=[(node_id=2, host='h31-x-251-59.host.redstation.co.uk', port=9092), (node_id=1, host='h31-x-251-58.host.redstation.co.uk', port=9092), (node_id=3, host='h31-x-251-60.host.redstation.co.uk', port=9092)], topics=[(error_code=0, topic='test-topic', partitions=[(error_code=0, partition=0, leader=1, replicas=[1, 3, 2], isr=[1, 3, 2])]), (error_code=0, topic='test-topic-666', partitions=[(error_code=0, partition=2, leader=3, replicas=[3, 1, 2], isr=[1, 3, 2]), (error_code=0, partition=1, leader=2, replicas=[2, 3, 1], isr=[1, 3, 2]), (error_code=0, partition=0, leader=1, replicas=[1, 2, 3], isr=[1, 3, 2])]), (error_code=0, topic='test-topic-444', partitions=[(error_code=0, partition=0, leader=1, replicas=[1], isr=[1])]), (error_code=0, topic='test-topic-21', partitions=[(error_code=0, partition=0, leader=3, replicas=[3], isr=[3])]), (error_code=0, topic='test-topic-777', partitions=[(error_code=0, partition=2, leader=3, replicas=[3, 1, 2], isr=[1, 3, 2]), (error_code=0, partition=1, leader=2, replicas=[2, 3, 1], isr=[1, 3, 2]), (error_code=0, partition=0, leader=1, replicas=[1, 2, 3], isr=[1, 3, 2])]), (error_code=0, topic='__consumer_offsets', partitions=[(error_code=0, partition=23, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=41, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=32, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=8, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=17, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=44, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=35, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=26, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=11, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=29, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=38, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=47, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=20, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=2, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=5, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=14, leader=1, replicas=[1], isr=[1]), (error_code=0, partition=46, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=49, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=40, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=4, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=13, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=22, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=31, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=16, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=7, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=43, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=25, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=34, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=10, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=37, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=1, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=19, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=28, leader=3, replicas=[3], isr=[3]), (error_code=0, partition=45, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=36, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=27, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=9, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=18, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=21, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=48, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=12, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=3, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=30, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=39, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=15, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=42, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=24, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=33, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=6, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=0, leader=2, replicas=[2], isr=[2])]), (error_code=0, topic='test-topic-2', partitions=[(error_code=0, partition=2, leader=1, replicas=[1, 2, 3], isr=[1, 3, 2]), (error_code=0, partition=1, leader=3, replicas=[3, 1, 2], isr=[1, 3, 2]), (error_code=0, partition=0, leader=2, replicas=[2, 3, 1], isr=[1, 3, 2])]), (error_code=0, topic='test-topic-222', partitions=[(error_code=0, partition=0, leader=2, replicas=[2], isr=[2])]), (error_code=0, topic='test-topic-333', partitions=[(error_code=0, partition=0, leader=2, replicas=[2], isr=[2])]), (error_code=0, topic='test-topic-555', partitions=[(error_code=0, partition=2, leader=3, replicas=[3, 1, 2], isr=[1, 3, 2]), (error_code=0, partition=1, leader=2, replicas=[2, 3, 1], isr=[1, 3, 2]), (error_code=0, partition=0, leader=1, replicas=[1, 2, 3], isr=[1, 3, 2])])])
DEBUG:kafka.cluster:Updated cluster metadata to ClusterMetadata(brokers: 3, topics: 10, groups: 0)
INFO:kafka.client:Bootstrap succeeded: found 3 brokers and 10 topics.
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: Closing connection. 
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=x.x.251.58/x.x.251.58 port=9092>: reconnect backoff 0.05913151426570893 after 1 failures
DEBUG:kafka.client:Initiating connection to node 2 at h31-x-251-59.host.redstation.co.uk:9092
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 node_id=2 host=h31-x-251-59.host.redstation.co.uk/h31-x-251-59.host.redstation.co.uk port=9092>: creating new socket
DEBUG:kafka.conn:<BrokerConnection node_id=2 host=h31-x-251-59.host.redstation.co.uk/x.x.251.59 port=9092>: setting socket option (6, 1, 1)
INFO:kafka.conn:<BrokerConnection node_id=2 host=h31-x-251-59.host.redstation.co.uk/x.x.251.59 port=9092>: connecting to x.x.251.59:9092
DEBUG:kafka.client:Node 2 connected
DEBUG:kafka.protocol.parser:Sending request ApiVersionRequest_v0()
DEBUG:kafka.protocol.parser:Sending request MetadataRequest_v0(topics=[])
DEBUG:kafka.protocol.parser:Received correlation id: 1
DEBUG:kafka.protocol.parser:Processing response ApiVersionResponse_v0
DEBUG:kafka.protocol.parser:Received correlation id: 2
DEBUG:kafka.protocol.parser:Processing response MetadataResponse_v0
INFO:kafka.conn:Broker version identifed as 1.0.0
INFO:kafka.conn:Set configuration api_version=(1, 0, 0) to skip auto check_version requests on startup
DEBUG:kafka.metrics.metrics:Added sensor with name bufferpool-wait-time
DEBUG:kafka.metrics.metrics:Added sensor with name batch-size
DEBUG:kafka.metrics.metrics:Added sensor with name compression-rate
DEBUG:kafka.metrics.metrics:Added sensor with name queue-time
DEBUG:kafka.metrics.metrics:Added sensor with name produce-throttle-time
DEBUG:kafka.metrics.metrics:Added sensor with name records-per-request
DEBUG:kafka.metrics.metrics:Added sensor with name bytes
DEBUG:kafka.metrics.metrics:Added sensor with name record-retries
DEBUG:kafka.metrics.metrics:Added sensor with name errors
DEBUG:kafka.metrics.metrics:Added sensor with name record-size-max
DEBUG:kafka.producer.sender:Starting Kafka producer I/O thread.
DEBUG:kafka.producer.kafka:Kafka producer started
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
INFO:kafka.client:Bootstrapping cluster metadata from [('localhost', 9092, <AddressFamily.AF_UNSPEC: 0>)]
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 node_id=bootstrap host=localhost/localhost port=9092>: creating new socket
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: setting socket option (6, 1, 1)
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: connecting to 127.0.0.1:9092
ERROR:kafka.conn:Connect attempt to <BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092> returned error 111. Disconnecting.
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: Closing connection. ConnectionError: 111
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: reconnect backoff 0.05293227544886118 after 1 failures
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: Closing connection. 
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=localhost/127.0.0.1 port=9092>: reconnect backoff 0.11365332715002299 after 2 failures
ERROR:kafka.client:Unable to bootstrap from [('localhost', 9092, <AddressFamily.AF_UNSPEC: 0>)]
Traceback (most recent call last):
  File "producer.py", line 19, in <module>
    producer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/producer/kafka.py", line 363, in __init__
    **self.config)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 234, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 836, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable
INFO:kafka.producer.kafka:Closing the Kafka producer with 0 secs timeout.
INFO:kafka.producer.kafka:Proceeding to force close the producer since pending requests could not be completed within timeout 0.
DEBUG:kafka.producer.sender:Beginning shutdown of Kafka producer I/O thread, sending remaining records.
INFO:kafka.conn:<BrokerConnection node_id=2 host=h31-x-251-59.host.redstation.co.uk/x.x.251.59 port=9092>: Closing connection. 
DEBUG:kafka.conn:<BrokerConnection node_id=2 host=h31-x-251-59.host.redstation.co.uk/x.x.251.59 port=9092>: reconnect backoff 0.044333261227126386 after 1 failures
DEBUG:kafka.producer.sender:Shutdown of Kafka producer I/O thread has completed.
DEBUG:kafka.producer.kafka:The Kafka producer has closed.
INFO:kafka.producer.kafka:Kafka producer closed
INFO:kafka.producer.kafka:Kafka producer closed

[dpkp]

It looks like your first producer succeeds connecting to a remote cluster but your second producer is trying to connect to localhost and fails. Are you intending to connect a second producer to localhost there?

[dpkp]

Specifically, this line suggests that you are missing a bootstrap_servers config and are inadvertently attempting to connect to localhost:

Traceback (most recent call last):
  File "producer.py", line 19, in <module>
    producer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))

[abhishek-bodapati]

hi i using: kafka-python :V 1.4.1 kafka broker : V 2.12-1.0.0 when run this code to connect to remote server:

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='x.x.251.60:9092')
producer.send('test-topic-222', b'some_message_bytes')
producer.flush()

I get this error :

Traceback (most recent call last):
  File "/var/www/html/kafka_python/producer.py", line 2, in <module>
    producer = KafkaProducer(bootstrap_servers='x.x.251.60:9092')
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/producer/kafka.py", line 363, in __init__
    **self.config)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 234, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/var/www/html/kafka_python/env/lib/python3.5/site-packages/kafka/client_async.py", line 856, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

else I did not uncomment in #listeners=PLAINTEXT://:9092 in server.properties

Try adding the api_version, something like this...

producer = KafkaProducer(bootstrap_servers='x.x.251.60:9092', api_version=(1, 4, 1))