code: def _create_producer(self):
          
          if not self.closed:
              try:
                  self.logger.debug("Creating new kafka producer using brokers: " +
                                     str(self.settings['KAFKA_HOSTS']))
  
                  return KafkaProducer(bootstrap_servers=self.settings['KAFKA_HOSTS'],
                                       value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                                       retries=3,
                                       linger_ms=self.settings['KAFKA_PRODUCER_BATCH_LINGER_MS'],
                                       buffer_memory=self.settings['KAFKA_PRODUCER_BUFFER_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 producer.",
                                     {'ex': traceback.format_exc()})
                  raise
docs: Tries to establish a Kafka consumer connection