old_contents: import logging
  from kafka_influxdb.encoder.errors import EncoderError
  
  
  class ReaderAbstract(object):
      """
      A Kafka consumer based on kafka-python
      See: https://github.com/dpkp/kafka-python
      """
  
      # We need to treat legacy Kafka Versions (< 0.9.0) a little different.
      # First, they don't work without Zookeeper and confluent-kafka has no support for Zookeeper.
      # Second, they don't support API discovery.
      # The specific workarounds required are documented below.
      KAFKA_VERSION_ZOOKEEPER_OPTIONAL = "0.9.0"
  
      # TODO: UNDO
      # def __init__(self, host, port, group, topic, broker_version=KAFKA_VERSION_ZOOKEEPER_OPTIONAL):
      def __init__(self, host, port, group, topic, broker_version="0.8.2"):
          """
          Initialize Kafka reader
          """
          self.host = host
          self.port = str(port)
          self.group = group
          self.topic = topic
          self.broker_version = broker_version
  
          # Initialized on read
          self.consumer = None
  
      def read(self):
          """
          Read from Kafka. Reconnect on error.
          """
          try:
              self._connect()
              for msg in self._handle_read():
                  yield msg
          finally:
              logging.info("Performing cleanup before stopping.")
              self._shutdown()
  
      def _connect(self):
          """
          Overwrite in child classes
          """
          raise NotImplementedError
  
      def _shutdown(self):
          """
          Cleanup tasks (e.g. closing the Kafka connection).
          Can be overwritten by specific readers if required.
          """
          self.consumer.close()
  
      def _handle_read(self):
          """
          Read messages from Kafka.
          Library-specific internal message handling.
          Needs to be implemented by every reader
          """
          raise NotImplementedError
  
new_contents: import logging
  from kafka_influxdb.encoder.errors import EncoderError
  
  
  class ReaderAbstract(object):
      """
      A Kafka consumer based on kafka-python
      See: https://github.com/dpkp/kafka-python
      """
  
      # We need to treat legacy Kafka Versions (< 0.9.0) a little different.
      # First, they don't work without Zookeeper and confluent-kafka has no support for Zookeeper.
      # Second, they don't support API discovery.
      # The specific workarounds required are documented below.
      KAFKA_VERSION_ZOOKEEPER_OPTIONAL = "0.9.0"
  
      def __init__(self, host, port, group, topic, broker_version=KAFKA_VERSION_ZOOKEEPER_OPTIONAL):
          """
          Initialize Kafka reader
          """
          self.host = host
          self.port = str(port)
          self.group = group
          self.topic = topic
          self.broker_version = broker_version
  
          # Initialized on read
          self.consumer = None
  
      def read(self):
          """
          Read from Kafka. Reconnect on error.
          """
          try:
              self._connect()
              for msg in self._handle_read():
                  yield msg
          finally:
              logging.info("Performing cleanup before stopping.")
              self._shutdown()
  
      def _connect(self):
          """
          Overwrite in child classes
          """
          raise NotImplementedError
  
      def _shutdown(self):
          """
          Cleanup tasks (e.g. closing the Kafka connection).
          Can be overwritten by specific readers if required.
          """
          if self.consumer:
              self.consumer.close()
  
      def _handle_read(self):
          """
          Read messages from Kafka.
          Library-specific internal message handling.
          Needs to be implemented by every reader
          """
          raise NotImplementedError
  
subject: Check if we actually have a consumer before closing it in the cleanup method.