async def METHOD_NAME(task_run):
    future = PrefectFuture(
        key=str(task_run.id),
        name="foo",
        task_runner=None,
        _final_state=Failed(data="foo"),
    )
    future.task_run = task_run
    future._submitted.set()
    def failed_fun():
        yield future
    assert await resolve_futures_to_data(failed_fun(), raise_on_failure=False) == []