[Regression?] MapStreamAsyncServer not streaming

[niall-nai]

Describe the bug
As the title says, MapStreamAsyncServer doesn't seem to be streaming data out as it's received. This seems to be a regression; a pipeline that was streaming around October last year (on pynumaflow v0.9.0 and numaflow v1.4.2) is now waiting for all processing to be completed before outputting- this is a problem for sufficiently large jobs as data is stored in memory until outputted

We've managed to trace this to the MapStreamAsyncServer call - running an equivalent function via the asyncio library instead of via pynumaflow functions results in messages being yielded as expected.

In particular this seems to be related to the underlying gRPC EoT call not being respected; during the course of recovering from this issue we cleared the isbsvc (in effect clearing the incoming buffer), and messages were yielded as expected

To Reproduce
Steps to reproduce the behavior:

1. Create a basic source->async map stream->sink pipeline
2. Start a job which will cause significant memory inflation - in our case, this is one which downloads a file and splits it into chunks. These chunks should be yielded and therefore pushed onto the sink as they're created
3. Note that the stream UDF grows in memory usage during processing (indicating data is being "stored")
4. Note that the stream UDF reduces in memory usage once processing has finished (indicating data is being outputted)

In our case specifically we run the pipeline in two threads, one which has a dequeue storing messages, and the other (main thread) which poplefts messages from that dequeue and yields them

Messages are not being sent to the sink at all (buffer shows 0 messages, sink shows a processing rate of 0), indicating they are being held by the server call. Nor is this a case of a yield being blocked (as searches online might suggest), as a logging call immediately after the yield is being respected

Expected behavior
Messages which are yielded before processing completes are passed on

Environment (please complete the following information):

• Numaflow: [v1.7.0]
• Numaflow-python: [v0.12.1]

---

Message from the maintainers:

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

[BulkBeing]

I'm checking this. @niall-nai would it be possible for you to provide a minimal example code to reproduce the issue ?

Also, if possible please try using the current main. A lot of refactoring and fixes were merged recently.

# uv
uv add "pynumaflow @ git+https://github.com/numaproj/numaflow-python.git@f894c01173a207a88adfea6b2b3dc8de71369ebc#subdirectory=packages/pynumaflow"

# or poetry
poetry add "git+https://github.com/numaproj/numaflow-python.git@f894c01173a207a88adfea6b2b3dc8de71369ebc#subdirectory=packages/pynumaflow"

[niall-nai]

Hi @BulkBeing it probably won't be today due to other commitments, but I can definitely look at getting that sorted for tomorrow, thanks for looking into this

[BulkBeing]

Thanks!
Meanwhile, I will also try to reproduce the issue from my side.

[niall-nai]

Just for completeness, I should note that we've ruled out the following so far:

• Hitting a thread limit (setting max_threads to 4 gives same result as 16)
• Hitting a CPU limit (sitting tidy on ~0.2vcpu)
• Issues caused by kubectl evictions vs a defined memory limit (setting memory limit to 500Mi on a task which was hitting 1.3Gi hit the same issue)
• Use of a dequeue possibly causing thread locking issues (switching to a (unreliable) version with list instead of dequeue had the same issues)

[kohlisid]

Just to confirm for my understanding @niall-nai, you mean instead of streaming they are doing a wait on the whole batch before releasing the messages to the next vertex right?

[niall-nai]

Exactly that @kohlisid, only when the handler function completes are any yielded messages actually leaving the python process