'Close this producer.
Arguments:
timeout (float, optional): timeout in seconds to wait for completion.'

def close(self, timeout=None):

    self._unregister_cleanup()
    if ((not hasattr(self, '_closed')) or self._closed):
        log.info('Kafka    producer    closed')
        return
    if (timeout is None):
        timeout = getattr(threading, 'TIMEOUT_MAX', 999999999)
    if getattr(threading, 'TIMEOUT_MAX', False):
        assert (0 <= timeout <= getattr(threading, 'TIMEOUT_MAX'))
    else:
        assert (timeout >= 0)
    log.info('Closing    the    Kafka    producer    with    %s    secs    timeout.', timeout)
    invoked_from_callback = bool((threading.current_thread() is self._sender))
    if (timeout > 0):
        if invoked_from_callback:
            log.warning('Overriding    close    timeout    %s    secs    to    0    in    order    to    prevent    useless    blocking    due    to    self-join.    This    means    you    have    incorrectly    invoked    close    with    a    non-zero    timeout    from    the    producer    call-back.', timeout)
        elif (self._sender is not None):
            self._sender.initiate_close()
            self._sender.join(timeout)
    if ((self._sender is not None) and self._sender.is_alive()):
        log.info('Proceeding    to    force    close    the    producer    since    pending    requests    could    not    be    completed    within    timeout    %s.', timeout)
        self._sender.force_close()
        if (not invoked_from_callback):
            self._sender.join()
    self._metrics.close()
    try:
        self.config['key_serializer'].close()
    except AttributeError:
        pass
    try:
        self.config['value_serializer'].close()
    except AttributeError:
        pass
    self._closed = True
    log.debug('The    Kafka    producer    has    closed.')
