**Description**
In its current shape the ExternalTaskSensor accepts either a single task_id or None to poll for the completion of a dag run. We have a use case where a dependent dag should poll for only certain list of tasks in the upstream dag. One option is to add N ExternalTaskSensor nodes if there are N nodes to be dependent on but those will be too many Sensor Nodes in the dag and can be avoided if the ExternalTaskSensor can accept a list of task_ids to poll for. 


**Use case / motivation**

We have a upstream dag that updates a list of hive tables that is further used by a lot of downstream dags. This dag updates 100s of hive tables but some of the downstream dag depends only upon 10-20 of these tables, There are multiple dags which depends upon varied list of hive tables from the upstream dag. 


**Are you willing to submit a PR?**

Yes, we are willing to submit a PR for this

**Related Issues**

Not that I am aware of . I did a search on the issue list. 

airflow/sensors/external_task.py
total_states = set(total_states)
        if set(self.failed_states).intersection(set(self.allowed_states)):
            raise AirflowException(
                "Duplicate values provided as allowed "
                "`{}` and failed states `{}`".format(self.allowed_states, self.failed_states)
            )
        if external_task_id:
            if not total_states <= set(State.task_states):
                raise ValueError(
                    f'Valid values for `allowed_states` and `failed_states` '
                    f'when `external_task_id` is not `None`: {State.task_states}'
                )
        elif not total_states <= set(State.dag_states):
            raise ValueError(
                f'Valid values for `allowed_states` and `failed_states` '
                f'when `external_task_id` is `None`: {State.dag_states}'
            )
        if execution_delta is not None and execution_date_fn is not None:
            raise ValueError(
                'Only one of `execution_delta` or `execution_date_fn` may '
                'be provided to ExternalTaskSensor; not both.'
            )
        self.execution_delta = execution_delta
        self.execution_date_fn = execution_date_fn
        self.external_dag_id = external_dag_id
        self.external_task_id = external_task_id
        self.check_existence = check_existence
        self._has_checked_existence = False
    @provide_session
    def poke(self, context, session=None):