Open KunTjz opened 6 years ago
I have also found that kafkaConsumer iterator ignores consumer_timeout_ms if consumer assigns to a non exist topic. It seems to the same question.
Is this a bug? Or my configuration is wrong?
@dpkp thanks a lot
I experience the same thing. And the loop makes CPU go to 100%.
@RechardTheGrey I can't reproduce this with the latest release. Have you tried upgrading?
I am experiencing the same problem, and here is the step to reproduce:
First, save this piece of code to test_error.py
:
from kafka import KafkaProducer, KafkaConsumer
import time
config = {
'bootstrap_servers': 'localhost:9092',
'group_id': 'test',
'consumer_timeout_ms': 1000
}
consumer = KafkaConsumer('pdd-sec-precise-waf-result', **config)
while True:
for message in consumer:
print(message)
print('no message')
Second, start a kafka server, create a topic named test
, and run the test_error.py
, we should see there are "no message"
messages rolling in the terminal every 1 second.
Third, kill the kafka server. Turn back to your terminal, you will see there is no more no message
output.
So I followed the code of kafka-python, and finally I stepped into here. I found the fact that the only condition to break the infinite loop is to have the future variable set to None or change the future.is_done to True, but there is not any code to modify the future variable.
Maybe the call stack would help you:
poll (/home/jm/tmp/test/kafka/client_async.py:544)
ensure_coordinator_ready (/home/jm/tmp/test/kafka/coordinator/base.py:260)
poll (/home/jm/tmp/test/kafka/coordinator/consumer.py:262)
_message_generator (/home/jm/tmp/test/kafka/consumer/group.py:1053)
__next__ (/home/jm/tmp/test/kafka/consumer/group.py:1119)
<module> (/home/jm/tmp/test/test_error.py:23)
And thank you for this great opensource project, you are doing a great job.
After testing, I found a surprise. It will block, until the cluster is back. But I still want to know if there is any way to know if the cluster is down in consumer.
Hi! Forgive my poor English, but I need your help. @dpkp While stress testing Kafka, I have found that comsumer can block forever when all brokers down. After all brokers down, comsumer will fall into a loop to getting metadata from broker. This loop will never break util one or more brokers restart.
Is there a way that consumer can break this loop? I do not want block here, I want to do some other things when brokers down.
my code is like this:
After consumer block, I got this when keyboardInterrupt:![image](https://user-images.githubusercontent.com/5518995/33793815-d7d30ab2-dcf9-11e7-8bc9-e4f053cb3db0.png)
other infos: kafka_python-1.3.2 python2.6.3 kafka_2.11-0.10.2.1