'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.ConnectionError.'

def close(self, error=None):

    if (self.state is ConnectionStates.DISCONNECTED):
        if (error is not None):
            log.warning('%s:    close()    called    on    disconnected    connection    with    error:    %s', self, error)
            traceback.print_stack()
        return
    log.info('%s:    Closing    connection.    %s', self, (error or ''))
    self.state = ConnectionStates.DISCONNECTING
    self.config['state_change_callback'](self)
    self._update_reconnect_backoff()
    if self._sock:
        self._sock.close()
        self._sock = None
    self.state = ConnectionStates.DISCONNECTED
    self.last_attempt = time.time()
    self._sasl_auth_future = None
    self._receiving = False
    self._next_payload_bytes = 0
    self._rbuffer.seek(0)
    self._rbuffer.truncate()
    if (error is None):
        error = Errors.Cancelled(str(self))
    while self.in_flight_requests:
        ifr = self.in_flight_requests.popleft()
        ifr.future.failure(error)
    self.config['state_change_callback'](self)
