### Apache Airflow version

2.4.2

### What happened

There is a mapped task is getting marked as `upstream_failed` when none of its upstream tasks are `failed` or `upstream_failed`.

![image](https://user-images.githubusercontent.com/10802053/199615824-c44e7e2f-ed05-49c6-a225-e054836466e1.png)

In the above graph view, if `first_task` finishes before `second_task`, `first_task` immediately tries to expand `middle_task`. **Note - this is an important step to reproduce - The order the tasks finish matter.**

Note that the value of the Airflow configuration variable [`schedule_after_task_execution`](https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#schedule-after-task-execution) must be `True` (the default) for this to occur.

The expansion occurs when the Task supervisor performs the "mini scheduler", in [this line in `dagrun.py`](https://github.com/apache/airflow/blob/63638cd2162219bd0a67caface4b1f1f8b88cc10/airflow/models/dagrun.py#L749).

Which then marks `middle_task` as `upstream_failed` in [this line in `mappedoperator.py`](https://github.com/apache/airflow/blob/63638cd2162219bd0a67caface4b1f1f8b88cc10/airflow/models/mappedoperator.py#L652):

```
                # If the map length cannot be calculated (due to unavailable
                # upstream sources), fail the unmapped task.
```

I believe this was introduced by the PR [Fail task if mapping upstream fails](https://github.com/apache/airflow/pull/25757).

### What you think should happen instead

The dynamic tasks should successfully execute. I don't think the mapped task should expand because its upstream task hasn't completed at the time it's expanded. If the upstream task were to complete earlier, it would expand successfully.

### How to reproduce

Execute this DAG, making sure Airflow configuration `schedule_after_task_execution` is set to default value `True`.

```
from datetime import datetime, timedelta
import time

from airflow import DAG, XComArg
from airflow.operators.python import PythonOperator


class PrintIdOperator(PythonOperator):
    def __init__(self, id, **kwargs) -> None:
        super().__init__(**kwargs)
        self.op_kwargs["id"] = id


DAG_ID = "test_upstream_failed_on_mapped_operator_expansion"


default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "retry_delay": timedelta(minutes=1),
    "retries": 0,
}


def nop(id):
    print(f"{id=}")


def get_ids(delay: int = 0):
    print(f"Delaying {delay} seconds...")
    time.sleep(delay)
    print("Done!")
    return [0, 1, 2]


with DAG(
    dag_id=DAG_ID,
    default_args=default_args,
    start_date=datetime(2022, 8, 3),
    catchup=False,
    schedule=None,
    max_active_runs=1,
) as dag:

    second_task = PythonOperator(
        task_id="second_task",
        python_callable=get_ids,
        op_kwargs={"delay": 10}
    )

    first_task = PythonOperator(
        task_id="first_task",
        python_callable=get_ids,
    )

    middle_task = PrintIdOperator.partial(
        task_id="middle_task",
        python_callable=nop,
    ).expand(id=XComArg(second_task))

    last_task = PythonOperator(
        task_id="last_task",
        python_callable=nop,
        op_kwargs={"id": 1},
    )

    [first_task, middle_task] >> last_task

```

### Operating System

debian buster

### Versions of Apache Airflow Providers

_No response_

### Deployment

Docker-Compose

### Deployment details

_No response_

### Anything else

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""
        Ensure failure callback of a task is run by the airflow run --raw process
        """
        callback_file = tmp_path.joinpath("callback.txt")
        callback_file.touch()
        monkeypatch.setenv("AIRFLOW_CALLBACK_FILE", str(callback_file))
        dag = get_test_dag("test_on_failure_callback")
        with create_session() as session:
            dag.create_dagrun(
                state=State.RUNNING,
                execution_date=DEFAULT_DATE,
                run_type=DagRunType.SCHEDULED,
                session=session,
            )
        task = dag.get_task(task_id="test_on_failure_callback_task")
        ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
        ti.refresh_from_db()
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
        job1.run()
        ti.refresh_from_db()
        assert ti.state == State.FAILED  
        with open(callback_file) as f:
            lines = f.readlines()
        assert len(lines) == 1  
        assert lines[0].startswith(ti.key.primary)
        m = re.match(r"^.+pid: (\d+)$", lines[0])
        assert m, "pid expected in output."
        assert os.getpid() != int(m.group(1))
    def test_mark_success_on_success_callback(self, caplog, get_test_dag):