docs: def close(self, timeout=None):
          """Close this producer.
  
          Arguments:
              timeout (float, optional): timeout in seconds to wait for completion.
          """
  
          # drop our atexit handler now to avoid leaks
          self._unregister_cleanup()
  
          if not hasattr(self, '_closed') or self._closed:
              log.info('Kafka producer closed')
              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("Closing the Kafka producer with %s secs timeout.", timeout)
          #first_exception = AtomicReference() # this will keep track of the first encountered exception
          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)
              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("Proceeding to force close the producer since pending"
                       " requests could not be completed within timeout %s.",
                       timeout)
              self._sender.force_close()
              # Only join the sender thread when not calling from callback.
              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.")