'Manually assign a list of TopicPartitions to this consumer.
This interface does not allow for incremental assignment and will
replace the previous assignment (if there was one).
Manual topic assignment through this method does not use the consumer\'s
group management functionality. As such, there will be no rebalance
operation triggered when group membership or cluster and topic metadata
change. Note that it is not possible to use both manual partition
assignment with assign() and group assignment with subscribe().
Arguments:
partitions (list of TopicPartition): assignment for this instance.
Raises:
IllegalStateError: if consumer has already called subscribe()'

def assign_from_user(self, partitions):

    if (self.subscription is not None):
        raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
    self._user_assignment.clear()
    self._user_assignment.update(partitions)
    for partition in partitions:
        if (partition not in self.assignment):
            self._add_assigned_partition(partition)
    for tp in (set(self.assignment.keys()) - self._user_assignment):
        del self.assignment[tp]
    self.needs_partition_assignment = False
    self.needs_fetch_committed_offsets = True
