human
Complete this python program:
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):
assistant
#Here is the Python code you requested.

    """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
    :meth:`join`!
    Arguments:
    queue (threading.Queue): the queue from which to get messages
    client (kafka.SimpleClient): 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))):
        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)
        if request_tries:
            count = 0
            log.debug('Skipping    new    batch    collection    to    handle    retries')
        else:
            log.debug('Batching    size:    %s,    timeout:    %s', count, timeout)
        while ((count > 0) and (timeout >= 0)):
            try:
                (topic_partition, msg, key) = queue.get(timeout=timeout)
            except Empty:
                break
            if (topic_partition == STOP_ASYNC_PRODUCER):
                stop_event.set()
                break
            count -= 1
            timeout = (send_at - time.time())
            msgset[topic_partition].append((msg, key))
        for (topic_partition, msg) in msgset.items():
            messages = create_message_set(msg, codec, key, codec_compresslevel)
            req = ProduceRequestPayload(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):
    ...