code: def close(self):
          
          # If we're already waiting on an/some outstanding disconnects
          # make sure we continue to wait for them...
          log.debug("%r: close", self)
          self._closing = True
          # Close down any clients we have
          brokerclients, self.clients = self.clients, None
          self._close_brokerclients(brokerclients.values())
          # clean up other outstanding operations
          self.reset_all_metadata()
          return self.close_dlist or defer.succeed(None)
docs: Permanently dispose of the client
  
          - Immediately mark the client as closed, causing current operations to
            fail with :exc:`~afkak.common.CancelledError` and future operations to
            fail with :exc:`~afkak.common.ClientError`.
          - Clear cached metadata.
          - Close any connections to Kafka brokers.
  
          :returns:
              deferred that fires when all resources have been released