def assign(cls, cluster, members):
        """Performs group assignment given cluster metadata and member subscriptions

        Arguments:
            cluster (ClusterMetadata): cluster metadata
            members (dict of {member_id: MemberMetadata}): decoded metadata for each member in the group.

        Returns:
          dict: {member_id: MemberAssignment}
        """
        members_metadata = {}
        for consumer, member_metadata in six.iteritems(members):
            members_metadata[consumer] = cls.parse_member_metadata(member_metadata)

        executor = StickyAssignmentExecutor(cluster, members_metadata)
        executor.perform_initial_assignment()
        executor.balance()

        cls._latest_partition_movements = executor.partition_movements

        assignment = {}
        for member_id in members:
            assignment[member_id] = ConsumerProtocolMemberAssignment(
                cls.version, sorted(executor.get_final_assignment(member_id)), b''
            )
        return assignment
Performs group assignment given cluster metadata and member subscriptions

        Arguments:
            cluster (ClusterMetadata): cluster metadata
            members (dict of {member_id: MemberMetadata}): decoded metadata for each member in the group.

        Returns:
          dict: {member_id: MemberAssignment}
        
kafka/coordinator/assignors/sticky/sticky_assignor.py