'Manually assign a list of TopicPartitions to this consumer.
Arguments:
partitions (list of TopicPartition): Assignment for this instance.
Raises:
IllegalStateError: If consumer has already called
:meth:`~kafka.KafkaConsumer.subscribe`.
Warning:
It is not possible to use both manual partition assignment with
:meth:`~kafka.KafkaConsumer.assign` and group assignment with
:meth:`~kafka.KafkaConsumer.subscribe`.
Note:
This interface does not support incremental assignment and will
replace the previous assignment (if there was one).
Note:
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.'

def assign(self, partitions):

    self._subscription.assign_from_user(partitions)
    self._client.set_topics([tp.topic for tp in partitions])
