def publish(self, topic, messages, key=None, timeout=<NUM_LIT:2>):
if not isinstance(messages, list):<EOL><INDENT>messages = [messages]<EOL><DEDENT>try:<EOL><INDENT>for m in messages:<EOL><INDENT>m = encodeutils.safe_encode(m, incoming='<STR_LIT:utf-8>')<EOL>self._producer.produce(topic, m, key,<EOL>callback=KafkaProducer.delivery_report)<EOL>self._producer.poll(<NUM_LIT:0>)<EOL><DEDENT>return self._producer.flush(timeout)<EOL><DEDENT>except (BufferError, confluent_kafka.KafkaException,<EOL>NotImplementedError):<EOL><INDENT>log.exception(u'<STR_LIT>'.format(topic))<EOL>raise<EOL><DEDENT>
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