docs: def close(self, error=None):
          """Close socket and fail all in-flight-requests.
  
          Arguments:
              error (Exception, optional): pending in-flight-requests
                  will be failed with this exception.
                  Default: kafka.errors.KafkaConnectionError.
          """
          if self.state is ConnectionStates.DISCONNECTED:
              return
          with self._lock:
              if self.state is ConnectionStates.DISCONNECTED:
                  return
              log.info('%s: Closing connection. %s', self, error or '')
              self._update_reconnect_backoff()
              self._sasl_auth_future = None
              self._protocol = KafkaProtocol(
                  client_id=self.config['client_id'],
                  api_version=self.config['api_version'])
              if error is None:
                  error = Errors.Cancelled(str(self))
              ifrs = list(self.in_flight_requests.items())
              self.in_flight_requests.clear()
              self.state = ConnectionStates.DISCONNECTED
              # To avoid race conditions and/or deadlocks
              # keep a reference to the socket but leave it
              # open until after the state_change_callback
              # This should give clients a change to deregister
              # the socket fd from selectors cleanly.
              sock = self._sock
              self._sock = None
  
          # drop lock before state change callback and processing futures
          self.config['state_change_callback'](self.node_id, sock, self)
          sock.close()
          for (_correlation_id, (future, _timestamp)) in ifrs:
              future.failure(error)