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