(0, State.NONE),
        (1, State.NONE),
        (2, State.NONE),
        (3, State.NONE),
        (4, State.NONE),
    ]
    ti3 = dr.get_task_instance(task_id="task_2", map_index=3)
    ti3.task = task2
    ti3.state = TaskInstanceState.FAILED
    session.merge(ti3)
    session.flush()
    
    dr.verify_integrity()
    tis = dr.get_task_instances()
    indices = [(ti.map_index, ti.state) for ti in tis if ti.map_index >= 0]
    assert sorted(indices) == [
        (0, State.NONE),
        (1, State.NONE),
        (2, State.NONE),
        (3, TaskInstanceState.FAILED),
        (4, State.NONE),
    ]
def test_calls_to_verify_integrity_with_mapped_task_reduction_at_runtime(dag_maker, session):
    """
    Test reduction in mapped task at runtime with calls to dagrun.verify_integrity
    """
    from airflow.models import Variable
    Variable.set(key="arg1", value=[1, 2, 3])
    @task
    def task_1():