# -*- coding: utf-8 -*-

import logging
import time
from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer

class KafkaReader(object):

    def __init__(self, host, port, group, topic, connection_wait_time=2):
        """ Initialize Kafka reader """
        self.host = host
        self.port = port
        self.group = group
        self.topic = topic
        self.connection_wait_time = connection_wait_time

    def connect(self):
        connection = "{0}:{1}".format(self.host, self.port)
        logging.info("Connecting to Kafka at {}...", connection)
        kafka_client = KafkaClient(connection)
        self.consumer = SimpleConsumer(kafka_client,
                                       self.group,
                                       self.topic)

    def read(self):
        """ Yield messages from Kafka topic """
        while True:
            try:
                self.connect()
                for raw_message in self.consumer:
                    yield raw_message.message.value
            except Exception:
                logging.error("Connection to Kafka lost. Trying to reconnect to {}:{}...",
                                    self.host, self.port)
                time.sleep(self.connection_wait_time)

@@ -585,16 +585,21 @@
         
+self.
 kafka_cl
@@ -668,16 +668,21 @@
 onsumer(
+self.
 kafka_cl

Make kafka client a field in Reader