'Publish a message to a topic.
Arguments:
topic (str): topic where the message will be published
value (optional): message value. Must be type bytes, or be
serializable to bytes via configured value_serializer. If value
is None, key is required and message acts as a \'delete\'.
See kafka compaction documentation for more details:
http://kafka.apache.org/documentation.html#compaction
(compaction requires kafka >= 0.8.1)
partition (int, optional): optionally specify a partition. If not
set, the partition will be selected using the configured
\'partitioner\'.
key (optional): a key to associate with the message. Can be used to
determine which partition to send the message to. If partition
is None (and producer\'s partitioner config is left as default),
then messages with the same key will be delivered to the same
partition (but if key is None, partition is chosen randomly).
Must be type bytes, or be serializable to bytes via configured
key_serializer.
timestamp_ms (int, optional): epoch milliseconds (from Jan 1 1970 UTC)
to use as the message timestamp. Defaults to current time.
Returns:
FutureRecordMetadata: resolves to RecordMetadata
Raises:
KafkaTimeoutError: if unable to fetch topic metadata, or unable
to obtain memory buffer prior to configured max_block_ms'

def send(self, topic, value=None, key=None, partition=None, timestamp_ms=None):

    assert ((value is not None) or (self.config['api_version'] >= (0, 8, 1))), 'Null    messages    require    kafka    >=    0.8.1'
    assert (not ((value is None) and (key is None))), 'Need    at    least    one:    key    or    value'
    key_bytes = value_bytes = None
    try:
        self._wait_on_metadata(topic, (self.config['max_block_ms'] / 1000.0))
        key_bytes = self._serialize(self.config['key_serializer'], topic, key)
        value_bytes = self._serialize(self.config['value_serializer'], topic, value)
        partition = self._partition(topic, partition, key, value, key_bytes, value_bytes)
        message_size = (MessageSet.HEADER_SIZE + Message.HEADER_SIZE)
        if (key_bytes is not None):
            message_size += len(key_bytes)
        if (value_bytes is not None):
            message_size += len(value_bytes)
        self._ensure_valid_record_size(message_size)
        tp = TopicPartition(topic, partition)
        if (timestamp_ms is None):
            timestamp_ms = int((time.time() * 1000))
        log.debug('Sending    (key=%r    value=%r)    to    %s', key, value, tp)
        result = self._accumulator.append(tp, timestamp_ms, key_bytes, value_bytes, self.config['max_block_ms'])
        (future, batch_is_full, new_batch_created) = result
        if (batch_is_full or new_batch_created):
            log.debug('Waking    up    the    sender    since    %s    is    either    full    or    getting    a    new    batch', tp)
            self._sender.wakeup()
        return future
    except Errors.KafkaTimeoutError:
        raise
    except AssertionError:
        raise
    except Exception as e:
        log.debug('Exception    occurred    during    message    send:    %s', e)
        return FutureRecordMetadata(FutureProduceResult(TopicPartition(topic, partition)), (-1), None, None, (len(key_bytes) if (key_bytes is not None) else (-1)), (len(value_bytes) if (value_bytes is not None) else (-1))).failure(e)
