rohitgcs commented 7 years ago

Description

Producer doesn't connect and send messages. I send messages and it all runs, but when I try to flush, I get Connection Timed Out Errors.

How to reproduce

confluent_kafka.version() ('0.11.0', 720896) confluent_kafka.libversion() ('0.11.0', 721151) Kafka Broker 0.9.0.1

Full script: from confluent_kafka import Producer p = Producer({'bootstrap.servers': 'localhost', 'client.id': 'test-producer', 'debug': 'broker' ,'api.version.request': True}) for data in ['test'+str(i) for i in range(0,10)]: print data p.produce(KAFKA_TOPIC, data.encode('utf-8')) print "Before: ",len(p) time.sleep(2) print "After: ", len(p) p.poll(0)

p.flush()

p = Producer({'bootstrap.servers': 'localhost', 'client.id': 'test-producer', 'debug': 'broker' ,'api.version.request': True}) %7|1507675267.835|BRKMAIN|test-producer#producer-7| [thrd::0/internal]: :0/internal: Enter main broker thread %7|1507675267.835|BROKER|test-producer#producer-7| [thrd:app]: localhost:9092/bootstrap: Added new broker with NodeId -1 %7|1507675267.835|DESTROY|test-producer#producer-6| [thrd:app]: Terminating instance %7|1507675267.835|DESTROY|test-producer#producer-6| [thrd:main]: Destroy internal %7|1507675267.835|DESTROY|test-producer#producer-6| [thrd:main]: Removing all topics %7|1507675267.835|TERMINATE|test-producer#producer-6| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Handle is terminating: failed 0 request(s) in retry+outbuf %7|1507675267.835|BROKERFAIL|test-producer#producer-6| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: failed: err: Local: Broker handle destroyed: (errno: Operation now in progress) %7|1507675267.835|STATE|test-producer#producer-6| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN %7|1507675267.835|BRKMAIN|test-producer#producer-7| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Enter main broker thread %7|1507675267.835|CONNECT|test-producer#producer-7| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state INIT connecting %7|1507675267.835|CONNECT|test-producer#producer-7| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#localhost:9092 (plaintext) with socket 8 %7|1507675267.835|STATE|test-producer#producer-7| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> CONNECT %7|1507675267.835|STATE|test-producer#producer-7| [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP %7|1507675267.935|TERMINATE|test-producer#producer-6| [thrd::0/internal]: :0/internal: Handle is terminating: failed 0 request(s) in retry+outbuf %7|1507675267.935|BROKERFAIL|test-producer#producer-6| [thrd::0/internal]: :0/internal: failed: err: Local: Broker handle destroyed: (errno: Success) %7|1507675267.935|STATE|test-producer#producer-6| [thrd::0/internal]: :0/internal: Broker changed state UP -> DOWN

Checklist

Please provide the following information:

[x] confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()):
[X] Apache Kafka broker version:
[X] Client configuration: {...}
[ ] Operating system:
[X] Provide client logs (with 'debug': '..' as necessary)
[ ] Provide broker log excerpts
[ ] Critical issue

edenhill commented 7 years ago

Could you provide some more logs, the provided log only gives 100 milliseconds of output.

Side note: len(Producer) will return both messages queued for tranmission, in transit, and messages that have been delivered/failed delivery. The latter bunch will only get served/removed when you call poll() or flush().

Chloejay commented 4 years ago

same issue when rerun docker localhost:9092/bootstrap: failed: err: Local: Broker handle destroyed: (errno: Operation now in progress),

here is the logs

This is docker-compose ps

            Name                         Command            State                 Ports              
------------------------------------------------------------------------------------------------------
kafka_docker_connect_1           /etc/confluent/docker/run   Up       0.0.0.0:8083->8083/tcp, 9092/tcp
kafka_docker_control-center_1    /etc/confluent/docker/run   Up       0.0.0.0:9021->9021/tcp          
kafka_docker_kafka_1             /etc/confluent/docker/run   Exit 1                                   
kafka_docker_ksql-server_1       /etc/confluent/docker/run   Up       0.0.0.0:8088->8088/tcp          
kafka_docker_schema-registry_1   /etc/confluent/docker/run   Exit 1                                   
kafka_docker_tools_1             /bin/sh                     Up       8083/tcp, 9092/tcp              
kafka_docker_zookeeper_1         /etc/confluent/docker/run   Up       2181/tcp, 2888/tcp, 3888/tcp

It failed when running producer.py.

edenhill commented 4 years ago

Looks like you are not waiting for messages to be delivered before terminating the application. You need to use flush().

Chloejay commented 4 years ago

@edenhill thanks for the hint, solved it after your advice.

confluentinc / confluent-kafka-python

Producer fails to send messages #264

Description

How to reproduce

Checklist

here is the logs

This is docker-compose ps