Help: KafkaConsumer hangs forever when retrieve messages

[RoadToTop]

I wrote some python codes to retrieve Kafka messages from brokers. Below is my code.

def read_message(topic ,partition, broker, from_offset, until_offset):
    clientName = "Client_" + str(topic) + "_" + str(partition)
    consumer = KafkaConsumer(bootstrap_servers=broker, client_id=clientName)
    topicPartition = TopicPartition(topic,partition)
    consumer.assign([topicPartition])
    consumer.seek(topicPartition, from_offset)
    result_list = []
    for msg in consumer:
        if msg.offset >= until_offset:
            break
        else:
            result_list.append(msg)
    return result_list

"broker" here is the lead broker that I find in other method.
"partition" is the partition that messages should be in.
"from_offset" and "until_offset" is the offset range that I want to retrieve messages in.

However, after "result_list" is created, it hangs forever. This function used to work properly. But sometimes it hangs.
I don't know what is the problem. It works in most cases. However, it hangs sometimes.

[dpkp]

What do the logs say?

[RoadToTop]

Well. The problem is that there has no log returned. It just hang there forever. Maybe adding more logs into this package will be great helpful.

[dpkp]

try adding this:

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

[aswinjoseroy]

I face the same issue. Can't get the consumer to run without hanging at some point. No Exceptions raised by the consumer either.

[tkl-artyomk]

Guys make sure that:

• all brokers are resolvable by (DNS | /etc/hosts)
• consumer can access brokers on port (9092)
• hostnames passed to bootstrap_severs are same as configured in advertised.host.name (e.g. passing IP address will not work).

[pavelvelikhov]

I'm having the same issue, I have switched to SimpleConsumer and it works fine. KafkaConsumer hangs though.

Here is my tornado example:

 class RestHandler(tornado.web.RequestHandler):
   def initialize(self,brokers):
     self.client = SimpleClient(hosts=brokers)
     self.consumer = SimpleConsumer(topic='test2',group=None,client=self.client,auto_commit=False)
    self.producer = KafkaProducer(bootstrap_servers=brokers)
 
   def post(self,pipeline):
    data_json = tornado.escape.json_decode(self.request.body)
     message = json.dumps({'header':[],'body':data_json})
     self.producer.send('test2',json.dumps(message).encode('utf-8'))
 
     result = self.consumer.get_messages(1)[0].message.value
     result = json.loads(result.decode('utf-8'))
     self.write(json.dumps(result))
    self.finish()

Should I just keep using SimpleConsumer? I want to have fault tolerance, so would like the consumer to adapt to cluster node failures

[M0bil3Rulz]

Looks like you have to set a timeout:

consumer_timeout_ms (int) – number of milliseconds to block during message iteration before raising StopIteration (i.e., ending the iterator). Default block forever [float(‘inf’)].

So lets assume you have nothing on your topic currently, when you start your KafkaConsumer, it will sit there and wait for messages to come in (via a next iterator). Once messages comes in, your consumer will process those messages and then continue to wait.

Setting the consumer_timeout_ms to some value (like 10 seconds or so), will cause the KafkaConsumer to stop processing after X amount of time without receiving a new message. Not sure if that was the intent or not, but that works for me.

consumer = KafkaConsumer( args.topic, bootstrap_servers=args.bootstrap_servers.split(), auto_offset_reset=args.auto_offset_reset, enable_auto_commit=args.enable_auto_commit, group_id=args.group_id, value_deserializer=lambda x: x.decode('utf-8'), consumer_timeout_ms=10000)

(args is just grabbing info from python command line)