'Close the coordinator, leave the current group,
and reset local generation / member_id'

def close(self):

    try:
        self._client.unschedule(self.heartbeat_task)
    except KeyError:
        pass
    if ((not self.coordinator_unknown()) and (self.generation > 0)):
        log.info('Leaving    consumer    group    (%s).', self.group_id)
        request = LeaveGroupRequest[0](self.group_id, self.member_id)
        future = self._client.send(self.coordinator_id, request)
        future.add_callback(self._handle_leave_group_response)
        future.add_errback(log.error, 'LeaveGroup    request    failed:    %s')
        self._client.poll(future=future)
    self.generation = OffsetCommitRequest[2].DEFAULT_GENERATION_ID
    self.member_id = JoinGroupRequest[0].UNKNOWN_MEMBER_ID
    self.rejoin_needed = True
