class_docstr: The background thread that handles the sending of produce requests to the
  Kafka cluster. This thread makes metadata requests to renew its view of the
  cluster and then sends produce requests to the appropriate nodes.
docstr: Run a single iteration of sending.
code: 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(timeout_ms=poll_timeout_ms)
  
imports: from __future__ import absolute_import, division
  import collections
  import copy
  import logging
  import threading
  import time
  from kafka.vendor import six
  from kafka import errors
  from kafka.metrics.measurable import AnonMeasurable
  from kafka.metrics.stats import Avg, Max, Rate
  from kafka.protocol.produce import ProduceRequest
  from kafka.structs import TopicPartition
  from kafka.version import __version__
label_desc: Use image node_id 3 for calling the Sender obj's underlying member method code with example usage: obj.run_once() without return types