[aDAG][core] Multi arg DAG fails if it raises an exception.

[rkooo567]

What happened + What you expected to happen

Right now, when you have multi arg task, and if it raises an exception, it couldn't handle the error. Ideally, all the downstream refs should raise an exception in this case.

class A:
    @ray.method(num_returns=2)
    def f(inp):
        raise Exception
        return 1, 2

with InputNode() as inp:
    x, y = a.f.bind(inp)
    dag = MultioutputNode([x, y])

Versions / Dependencies

master

Reproduction script

see above

Issue Severity

None