docs: def assign_from_subscribed(self, assignments):
          """Update the assignment to the specified partitions
  
          This method is called by the coordinator to dynamically assign
          partitions based on the consumer's topic subscription. This is different
          from assign_from_user() which directly sets the assignment from a
          user-supplied TopicPartition list.
  
          Arguments:
              assignments (list of TopicPartition): partitions to assign to this
                  consumer instance.
          """
          if not self.partitions_auto_assigned():
              raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
  
          for tp in assignments:
              if tp.topic not in self.subscription:
                  raise ValueError("Assigned partition %s for non-subscribed topic." % (tp,))
  
          # after rebalancing, we always reinitialize the assignment state
          self.assignment.clear()
          for tp in assignments:
              self._add_assigned_partition(tp)
          self.needs_fetch_committed_offsets = True
          log.info("Updated partition assignment: %s", assignments)