code: def _create_consumer(self):
          
          if not self.closed:
              try:
                  self.logger.debug("Creating new kafka consumer using brokers: " +
                                     str(self.settings['KAFKA_HOSTS']) + ' and topic ' +
                                     self.settings['KAFKA_TOPIC_PREFIX'] +
                                     ".outbound_firehose")
  
                  return KafkaConsumer(
                      self.settings['KAFKA_TOPIC_PREFIX'] + ".outbound_firehose",
                      group_id=None,
                      bootstrap_servers=self.settings['KAFKA_HOSTS'],
                      consumer_timeout_ms=self.settings['KAFKA_CONSUMER_TIMEOUT'],
                      auto_offset_reset=self.settings['KAFKA_CONSUMER_AUTO_OFFSET_RESET'],
                      auto_commit_interval_ms=self.settings['KAFKA_CONSUMER_COMMIT_INTERVAL_MS'],
                      enable_auto_commit=self.settings['KAFKA_CONSUMER_AUTO_COMMIT_ENABLE'],
                      max_partition_fetch_bytes=self.settings['KAFKA_CONSUMER_FETCH_MESSAGE_MAX_BYTES'])
              except KeyError as e:
                  self.logger.error('Missing setting named ' + str(e),
                                     {'ex': traceback.format_exc()})
              except:
                  self.logger.error("Couldn't initialize kafka consumer for topic",
                                     {'ex': traceback.format_exc()})
                  raise
docs: Tries to establing the Kafka consumer connection