'Run a single iteration of sending.'

def run_once(self):

    while self._topics_to_add:
        self._client.add_topic(self._topics_to_add.pop())
    result = self._accumulator.ready(self._metadata)
    (ready_nodes, next_ready_check_delay, unknown_leaders_exist) = result
    if unknown_leaders_exist:
        log.debug('Unknown    leaders    exist,    requesting    metadata    update')
        self._metadata.request_update()
    not_ready_timeout = 999999999
    for node in list(ready_nodes):
        if (not self._client.ready(node)):
            log.debug('Node    %s    not    ready;    delaying    produce    of    accumulated    batch', node)
            ready_nodes.remove(node)
            not_ready_timeout = min(not_ready_timeout, self._client.connection_delay(node))
    batches_by_node = self._accumulator.drain(self._metadata, ready_nodes, self.config['max_request_size'])
    if self.config['guarantee_message_order']:
        for batch_list in six.itervalues(batches_by_node):
            for batch in batch_list:
                self._accumulator.muted.add(batch.topic_partition)
    expired_batches = self._accumulator.abort_expired_batches(self.config['request_timeout_ms'], self._metadata)
    for expired_batch in expired_batches:
        self._sensors.record_errors(expired_batch.topic_partition.topic, expired_batch.record_count)
    self._sensors.update_produce_request_metrics(batches_by_node)
    requests = self._create_produce_requests(batches_by_node)
    poll_timeout_ms = min((next_ready_check_delay * 1000), not_ready_timeout)
    if ready_nodes:
        log.debug('Nodes    with    data    ready    to    send:    %s', ready_nodes)
        log.debug('Created    %d    produce    requests:    %s', len(requests), requests)
        poll_timeout_ms = 0
    for (node_id, request) in six.iteritems(requests):
        batches = batches_by_node[node_id]
        log.debug('Sending    Produce    Request:    %r', request)
        self._client.send(node_id, request).add_callback(self._handle_produce_response, node_id, time.time(), batches).add_errback(self._failed_produce, batches, node_id)
    self._client.poll(poll_timeout_ms, sleep=True)
