code: def assign(self, partitions):
          
          self._subscription.assign_from_user(partitions)
          self._client.set_topics([tp.topic for tp in partitions])
docs: 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.