def stop(self, timeout=None):
        """
        Stop the producer (async mode). Blocks until async thread completes.
        """
        if timeout is not None:
            log.warning('timeout argument to stop() is deprecated - '
                        'it will be removed in future release')

        if not self.async:
            log.warning('producer.stop() called, but producer is not async')
            return

        if self.stopped:
            log.warning('producer.stop() called, but producer is already stopped')
            return

        if self.async:
            self.queue.put((STOP_ASYNC_PRODUCER, None, None))
            self.thread_stop_event.set()
            self.thread.join()

        if hasattr(self, '_cleanup_func'):
            # Remove cleanup handler now that we've stopped

            # py3 supports unregistering
            if hasattr(atexit, 'unregister'):
                atexit.unregister(self._cleanup_func) # pylint: disable=no-member

            # py2 requires removing from private attribute...
            else:

                # ValueError on list.remove() if the exithandler no longer exists
                # but that is fine here
                try:
                    atexit._exithandlers.remove(  # pylint: disable=no-member
                        (self._cleanup_func, (self,), {}))
                except __HOLE__:
                    pass

            del self._cleanup_func

        self.stopped = True
dataset/ETHPy150Open dpkp/kafka-python/kafka/producer/base.py/Producer.stop