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.

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)