docs: def run_once(self):
          """Run a single iteration of sending."""
          while self._topics_to_add:
              self._client.add_topic(self._topics_to_add.pop())
  
          # get the list of partitions with data ready to send
          result = self._accumulator.ready(self._metadata)
          ready_nodes, next_ready_check_delay, unknown_leaders_exist = result
  
          # if there are any partitions whose leaders are not known yet, force
          # metadata update
          if unknown_leaders_exist:
              log.debug('Unknown leaders exist, requesting metadata update')
              self._metadata.request_update()
  
          # remove any nodes we aren't ready to send to
          not_ready_timeout = float('inf')
          for node in list(ready_nodes):
              if not self._client.is_ready(node):
                  log.debug('Node %s not ready; delaying produce of accumulated batch', node)
                  self._client.maybe_connect(node, wakeup=False)
                  ready_nodes.remove(node)
                  not_ready_timeout = min(not_ready_timeout,
                                          self._client.connection_delay(node))
  
          # create produce requests
          batches_by_node = self._accumulator.drain(
              self._metadata, ready_nodes, self.config['max_request_size'])
  
          if self.config['guarantee_message_order']:
              # Mute all the partitions drained
              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)
          # If we have any nodes that are ready to send + have sendable data,
          # poll with 0 timeout so this can immediately loop and try sending more
          # data. Otherwise, the timeout is determined by nodes that have
          # partitions with data that isn't yet sendable (e.g. lingering, backing
          # off). Note that this specifically does not include nodes with
          # sendable data that aren't ready to send since they would cause busy
          # looping.
          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) # trace
              log.debug("Created %d produce requests: %s", len(requests), requests) # trace
              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, wakeup=False)
                   .add_callback(
                       self._handle_produce_response, node_id, time.time(), batches)
                   .add_errback(
                       self._failed_produce, batches, node_id))
  
          # if some partitions are already ready to be sent, the select time
          # would be 0; otherwise if some partition already has some data
          # accumulated but not ready yet, the select time will be the time
          # difference between now and its linger expiry time; otherwise the
          # select time will be the time difference between now and the
          # metadata expiry time
          self._client.poll(poll_timeout_ms)