**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
if self.execution_delta:
            dttm = context['execution_date'] - self.execution_delta
        elif self.execution_date_fn:
            dttm = self._handle_execution_date_fn(context=context)
        else:
            dttm = context['execution_date']
        dttm_filter = dttm if isinstance(dttm, list) else [dttm]
        serialized_dttm_filter = ','.join(dt.isoformat() for dt in dttm_filter)
        self.log.info(
            'Poking for %s.%s on %s ... ', self.external_dag_id, self.external_task_id, serialized_dttm_filter
        )
        
        if self.check_existence and not self._has_checked_existence:
            self._check_for_existence(session=session)
        count_allowed = self.get_count(dttm_filter, session, self.allowed_states)
        count_failed = -1
        if self.failed_states:
            count_failed = self.get_count(dttm_filter, session, self.failed_states)
        if count_failed == len(dttm_filter):
            if self.external_task_id:
                raise AirflowException(
                    f'The external task {self.external_task_id} in DAG {self.external_dag_id} failed.'
                )
            else:
                raise AirflowException(f'The external DAG {self.external_dag_id} failed.')
        return count_allowed == len(dttm_filter)
    def _check_for_existence(self, session) -> None: