refactor kafka_consumer.py