docs: def publish(self, topic, messages, key=None, timeout=2):
          """
          Publish messages to the topic.
  
          :param str topic: Topic to produce messages to.
          :param list(str) messages:  List of message payloads.
          :param str key: Message key.
          :param float timeout: Maximum time to block in seconds.
          :returns: Number of messages still in queue.
          :rtype int
          """
  
          if not isinstance(messages, list):
              messages = [messages]
  
          try:
              for m in messages:
                  m = encodeutils.safe_encode(m, incoming='utf-8')
                  self._producer.produce(topic, m, key,
                                         callback=KafkaProducer.delivery_report)
                  self._producer.poll(0)
  
              return self._producer.flush(timeout)
  
          except (BufferError, confluent_kafka.KafkaException,
                  NotImplementedError):
              log.exception(u'Error publishing to {} topic.'.format(topic))
              raise