docs: def producer(self):
          """
          :raises: kafka.errors.NoBrokersAvailable if the connection is broken
          """
          if self._producer:
              return self._producer
  
          self._producer = KafkaProducer(
              bootstrap_servers=self.hosts,
              value_serializer=lambda v: self._serializer_class().dumps(v).encode('utf-8'),
              **self._kwargs
          )
  
          return self._producer