**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. 

tests/sensors/test_external_task_sensor.py
task_external_with_failure = BashOperator(
            task_id="task_external_with_failure", bash_command=bash_command_code, retries=0, dag=dag_external
        )
        task_external_without_failure = DummyOperator(
            task_id="task_external_without_failure", retries=0, dag=dag_external
        )
        task_external_without_failure.run(
            start_date=DEFAULT_DATE, end_date=DEFAULT_DATE + timedelta(seconds=1), ignore_ti_state=True
        )
        session = settings.Session()
        TI = TaskInstance
        try:
            task_external_with_failure.run(
                start_date=DEFAULT_DATE, end_date=DEFAULT_DATE + timedelta(seconds=1), ignore_ti_state=True
            )
            
            
            
        except Exception as e:
            failed_tis = (
                session.query(TI)
                .filter(
                    TI.dag_id == dag_external_id,
                    TI.state == State.FAILED,
                    TI.execution_date == DEFAULT_DATE + timedelta(seconds=1),
                )
                .all()
            )
            if len(failed_tis) == 1 and failed_tis[0].task_id == 'task_external_with_failure':
                pass