Kafka Consumers stops receiving messages after few days

[vaibhav293]

We have 3 cluster prod and 1 node dev setup. We are observing the similar behaviour in both the setups.

My consumer code looks something like this:

from kafka import KafkaConsumer

consumer = KafkaConsumer("topic1","topic2"
                        bootstrap_servers=<my servers>,
                        api_version="0.10.1",
                        group_id="my_group")

for msg in consumer:
    # Print the index of message
    print(msg.offset)

    if message['event_name'] == "topic1":
            MyController().process_topic1(**message)
        elif message['event_name'] == "topic2":
            MyController().process_topic2(**message) 

We also run these consumers using supervisord as 3 processes to consume messages from 3 partitions.

When I restart all the consumers (supervisord processes) they start consuming messages. but after some days most of the consumers stop consuming messages.

$ ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my_group
Note: This will not show information about old Zookeeper-based consumers.

Consumer group 'my_group' has no active members.

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
topic1            0          139             139             0          -                                                 -                              -
topic2            0          84              87              3          -                                                 -                              -

I also see that kafka-consumer-groups.sh stops showing the consumer-ids for some of the groups.

The frequency of messages in the system is very less as of now.

Please suggest how should I investigate and troubleshoot this problem.

[dpkp]

What do the logs say? Try enabling debug logs if this is reproducible.

[vaibhav293]

I have not enabled the client side debug logging.

Something like this:
import logging
import sys
logger = logging.getLogger('kafka')
logger.addHandler(logging.StreamHandler(sys.stdout))
logger.setLevel(logging.DEBUG)

Will enable and share observations soon.

[andrewkowalik]

What version of kafka-python @vaibhav293 ?

@dpkp I am not saying it is the same but this feels really similar to the issue we have been noticing in production. I am going to dig more into the kafka-python timeout logic.

[vaibhav293]

@infecto kafka-python version 1.3.5

[jeffwidman]

@vaibhav293 please upgrade to the latest version and see if the problem still exists. There were a lot of changes in the 1.4.x series, and so IMHO not really worth the time to try to track this down in an old version