def close(self, timeout=None, null_logger=False):
        """Close this producer.

        Arguments:
            timeout (float, optional): timeout in seconds to wait for completion.
        """
        if null_logger:
            # Disable logger during destruction to avoid touching dangling references
            class NullLogger(object):
                def __getattr__(self, name):
                    return lambda *args: None

            global log
            log = NullLogger()

        # drop our atexit handler now to avoid leaks
        self._unregister_cleanup()

        if not hasattr(self, '_closed') or self._closed:
            log.info('%s: Kafka producer closed', str(self))
            return
        if timeout is None:
            # threading.TIMEOUT_MAX is available in Python3.3+
            timeout = getattr(threading, 'TIMEOUT_MAX', float('inf'))
        if getattr(threading, 'TIMEOUT_MAX', False):
            assert 0 <= timeout <= getattr(threading, 'TIMEOUT_MAX')
        else:
            assert timeout >= 0

        log.info("%s: Closing the Kafka producer with %s secs timeout.", str(self), timeout)
        self.flush(timeout)
        invoked_from_callback = bool(threading.current_thread() is self._sender)
        if timeout > 0:
            if invoked_from_callback:
                log.warning("%s: 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.",
                            str(self), timeout)
            else:
                # Try to close gracefully.
                if 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("%s: Proceeding to force close the producer since pending"
                     " requests could not be completed within timeout %s.",
                     str(self), timeout)
            self._sender.force_close()

        if self._metrics:
            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("%s: The Kafka producer has closed.", str(self))
Close this producer.

        Arguments:
            timeout (float, optional): timeout in seconds to wait for completion.
        
kafka/producer/kafka.py