docs: def assign_from_user(self, partitions):
          """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()
          """
          if self.subscription is not None:
              raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
  
          if self._user_assignment != set(partitions):
              self._user_assignment = set(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_fetch_committed_offsets = True