docs: def _send_upstream(queue, client, codec, batch_time, batch_size,
                     req_acks, ack_timeout, retry_options, stop_event,
                     log_messages_on_error=ASYNC_LOG_MESSAGES_ON_ERROR,
                     stop_timeout=ASYNC_STOP_TIMEOUT_SECS,
                     codec_compresslevel=None):
      """Private method to manage producing messages asynchronously
  
      Listens on the queue for a specified number of messages or until
      a specified timeout and then sends messages to the brokers in grouped
      requests (one per broker).
  
      Messages placed on the queue should be tuples that conform to this format:
          ((topic, partition), message, key)
  
      Currently does not mark messages with task_done. Do not attempt to join()!
  
      Arguments:
          queue (threading.Queue): the queue from which to get messages
          client (KafkaClient): instance to use for communicating with brokers
          codec (kafka.protocol.ALL_CODECS): compression codec to use
          batch_time (int): interval in seconds to send message batches
          batch_size (int): count of messages that will trigger an immediate send
          req_acks: required acks to use with ProduceRequests. see server protocol
          ack_timeout: timeout to wait for required acks. see server protocol
          retry_options (RetryOptions): settings for retry limits, backoff etc
          stop_event (threading.Event): event to monitor for shutdown signal.
              when this event is 'set', the producer will stop sending messages.
          log_messages_on_error (bool, optional): log stringified message-contents
              on any produce error, otherwise only log a hash() of the contents,
              defaults to True.
          stop_timeout (int or float, optional): number of seconds to continue
              retrying messages after stop_event is set, defaults to 30.
      """
      request_tries = {}
  
      while not stop_event.is_set():
          try:
              client.reinit()
          except Exception as e:
              log.warn(
                  'Async producer failed to connect to brokers; backoff for %s(ms) before retrying',
                  retry_options.backoff_ms)
              time.sleep(float(retry_options.backoff_ms) / 1000)
          else:
              break
  
      stop_at = None
      while not (stop_event.is_set() and queue.empty() and not request_tries):
  
          # Handle stop_timeout
          if stop_event.is_set():
              if not stop_at:
                  stop_at = stop_timeout + time.time()
              if time.time() > stop_at:
                  log.debug('Async producer stopping due to stop_timeout')
                  break
  
          timeout = batch_time
          count = batch_size
          send_at = time.time() + timeout
          msgset = defaultdict(list)
  
          # Merging messages will require a bit more work to manage correctly
          # for now, dont look for new batches if we have old ones to retry
          if request_tries:
              count = 0
              log.debug('Skipping new batch collection to handle retries')
          else:
              log.debug('Batching size: %s, timeout: %s', count, timeout)
  
          # Keep fetching till we gather enough messages or a
          # timeout is reached
          while count > 0 and timeout >= 0:
              try:
                  topic_partition, msg, key = queue.get(timeout=timeout)
              except Empty:
                  break
  
              # Check if the controller has requested us to stop
              if topic_partition == STOP_ASYNC_PRODUCER:
                  stop_event.set()
                  break
  
              # Adjust the timeout to match the remaining period
              count -= 1
              timeout = send_at - time.time()
              msgset[topic_partition].append((msg, key))
  
          # Send collected requests upstream
          for topic_partition, msg in msgset.items():
              messages = create_message_set(msg, codec, key, codec_compresslevel)
              req = ProduceRequest(topic_partition.topic,
                                   topic_partition.partition,
                                   tuple(messages))
              request_tries[req] = 0
  
          if not request_tries:
              continue
  
          reqs_to_retry, error_cls = [], None
          retry_state = {
              'do_backoff': False,
              'do_refresh': False
          }
  
          def _handle_error(error_cls, request):
              if (issubclass(error_cls, RETRY_ERROR_TYPES) or
                      (retry_options.retry_on_timeouts and
                       issubclass(error_cls, RequestTimedOutError))):
                  reqs_to_retry.append(request)
              if issubclass(error_cls, RETRY_BACKOFF_ERROR_TYPES):
                  retry_state['do_backoff'] |= True
              if issubclass(error_cls, RETRY_REFRESH_ERROR_TYPES):
                  retry_state['do_refresh'] |= True
  
          requests = list(request_tries.keys())
          log.debug('Sending: %s', requests)
          responses = client.send_produce_request(requests,
                                                  acks=req_acks,
                                                  timeout=ack_timeout,
                                                  fail_on_error=False)
  
          log.debug('Received: %s', responses)
          for i, response in enumerate(responses):
              error_cls = None
              if isinstance(response, FailedPayloadsError):
                  error_cls = response.__class__
                  orig_req = response.payload
  
              elif isinstance(response, ProduceResponse) and response.error:
                  error_cls = kafka_errors.get(response.error, UnknownError)
                  orig_req = requests[i]
  
              if error_cls:
                  _handle_error(error_cls, orig_req)
                  log.error('%s sending ProduceRequest (#%d of %d) '
                            'to %s:%d with msgs %s',
                            error_cls.__name__, (i + 1), len(requests),
                            orig_req.topic, orig_req.partition,
                            orig_req.messages if log_messages_on_error else hash(orig_req.messages))
  
          if not reqs_to_retry:
              request_tries = {}
              continue
  
          # doing backoff before next retry
          if retry_state['do_backoff'] and retry_options.backoff_ms:
              log.warn('Async producer backoff for %s(ms) before retrying', retry_options.backoff_ms)
              time.sleep(float(retry_options.backoff_ms) / 1000)
  
          # refresh topic metadata before next retry
          if retry_state['do_refresh']:
              log.warn('Async producer forcing metadata refresh metadata before retrying')
              try:
                  client.load_metadata_for_topics()
              except Exception as e:
                  log.error("Async producer couldn't reload topic metadata. Error: `%s`", e.message)
  
          # Apply retry limit, dropping messages that are over
          request_tries = dict(
              (key, count + 1)
              for (key, count) in request_tries.items()
              if key in reqs_to_retry and (retry_options.limit is None or
                                           (count < retry_options.limit))
          )
  
          # Log messages we are going to retry
          for orig_req in request_tries.keys():
              log.info('Retrying ProduceRequest to %s:%d with msgs %s',
                       orig_req.topic, orig_req.partition,
                       orig_req.messages if log_messages_on_error else hash(orig_req.messages))
  
      if request_tries or not queue.empty():
          log.error('Stopped producer with {0} unsent messages'
                    .format(len(request_tries) + queue.qsize()))