**Apache Airflow version**: 2.x including main

**What happened**:

I am using an `execution_date_fn` in an `ExternalTaskSensor` that also sets `allowed_states=['success']` and `failed_states=['failed']`. When one of the N upstream tasks fails, the sensor will hang forever in the `poke` method because there is a bug in checking for failed_states.

**What you expected to happen**:

I would expect the `ExternalTaskSensor` to fail.

I think this is due to a bug in the `poke` method where it should check if `count_failed > 0` as opposed to checking `count_failed == len(dttm_filter)`. I've created a fix locally that works for my case and have submitted a PR #16205 for it as reference.

**How to reproduce it**:

Create any `ExternalTaskSensor` that checks for `failed_states` and have one of the external DAGs tasks fail while others succeed.

E.g.
```
ExternalTaskSensor(
        task_id='check_external_dag',                                                     
        external_dag_id='external_dag',                                                               
        external_task_id=None,                              
        execution_date_fn=dependent_date_fn,
        allowed_states=['success'],
        failed_states=['failed'],                                                                     
        check_existence=True)
```

clear_db_runs()
@provide_session
def run_tasks(dag_bag, execution_date=DEFAULT_DATE, session=None):
    """
    Run all tasks in the DAGs in the given dag_bag. Return the TaskInstance objects as a dict
    keyed by task_id.
    """
    tis = {}
    for dag in dag_bag.dags.values():
        dagrun = dag.create_dagrun(
            state=State.RUNNING,
            execution_date=execution_date,
            start_date=execution_date,
            run_type=DagRunType.MANUAL,
            session=session,
        )
        
        
        
        
        tasks = sorted((ti for ti in dagrun.task_instances), key=lambda ti: ti.task_id)
        for ti in tasks:
            ti.refresh_from_task(dag.get_task(ti.task_id))
            tis[ti.task_id] = ti
            ti.run(session=session)
            session.flush()
            session.merge(ti)
            assert_ti_state_equal(ti, State.SUCCESS)
    return tis
def assert_ti_state_equal(task_instance, state):