def assign(self, partitions):
        """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.
        """
        if not partitions:
            self.unsubscribe()
        else:
            # make sure the offsets of topic partitions the consumer is unsubscribing from
            # are committed since there will be no following rebalance
            self._coordinator.maybe_auto_commit_offsets_now()
            self._subscription.assign_from_user(partitions)
            self._client.set_topics([tp.topic for tp in partitions])
            log.debug("Subscribed to partition(s): %s", partitions)
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.
        
kafka/consumer/group.py