def maybe_leave_group(self, timeout_ms=None):
        """Leave the current group and reset local generation/memberId."""
        if self.config['api_version'] < (0, 9):
            raise Errors.UnsupportedVersionError('Group Coordinator APIs require 0.9+ broker')
        with self._client._lock, self._lock:
            if (not self.coordinator_unknown()
                and self.state is not MemberState.UNJOINED
                and self._generation.is_valid):

                # this is a minimal effort attempt to leave the group. we do not
                # attempt any resending if the request fails or times out.
                log.info('Leaving consumer group (%s).', self.group_id)
                version = self._client.api_version(LeaveGroupRequest, max_version=2)
                request = LeaveGroupRequest[version](self.group_id, self._generation.member_id)
                log.debug('Sending LeaveGroupRequest to %s: %s', self.coordinator_id, request)
                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, timeout_ms=timeout_ms)

            self.reset_generation()
Leave the current group and reset local generation/memberId.
kafka/coordinator/base.py