[Data] Async MapBatches hangs upon exception raised from UDF

[scottjlee]

What happened + What you expected to happen

When using Ray Data's async map_batches API and an exception is raised from the UDF, the exception is not raised and the data execution will be hanging. See reproducible for a simple example

Versions / Dependencies

ray master b6d4792

Reproduction script

import ray

class MyUDF:
    def __init__(self):
        pass

    async def __call__(self, batch):
        assert False
        yield batch

ds = ray.data.range(20)
ds = ds.map_batches(MyUDF, concurrency=2)
result = ds.materialize()
assert len(result) == 20

We should expect the script above to run to completion, but instead hangs forever without any indication of exception or failure.

Issue Severity

None

[Bye-legumes]

@scottjlee I add a exception queue to fix it and this is what look like now.