Support for async UDFs for Map

[ab93]

Describe the bug gRPC async is not fully supported.

To Reproduce Steps to reproduce the behavior:

1. Define an async handler: async def map_handler(key: str, datum: Datum) -> Messages: val = datum.value _ = datum.event_time _ = datum.watermark messages = Messages(Message.to_vtx(key, val)) await asyncio.sleep(1) return messages

2. Run a simple forward message pipeline

Expected behavior The udf should work and just forward pass the message.

Screenshots

Environment (please complete the following information):

• Numaflow-python: v0.2.6

Message from the maintainers:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.

[vigith]

we have added first-class support for async map.