UDFs do not seem to propagate Watermarks

[niall-nai]

Describe the bug
UDFs created with this library do not seem to pass on watermarks to subsequent stages in the pipeline; this effectively makes Reduce components impossible to implement in anything but a simple pipeline

e.g: _event_time=datetime.datetime(2026, 3, 5, 17, 6, 34, 563000), _watermark=datetime.datetime(1969, 12, 31, 23, 59, 59, 999000), _headers={})

To Reproduce
Steps to reproduce the behavior:

1. Create a simple pipeline which has a source, map udf, reduce udf, and a sink
2. Log datums coming into the reduce UDF
3. Check the numaflow UI
4. Notice watermarks are set between source and map UDF
5. Notice watermarks are not set between map UDF and reduce UDF
6. Notice logged datums coming into reduce UDF have a set _event_time but _watermark is effectively set to -1

Expected behavior
Watermarks are propagated through UDFs

Screenshots
If applicable, add screenshots to help explain your problem.

Environment (please complete the following information):

• Kubernetes: v1.32.2
• Numaflow: v1.4.2 (also v1.7.3)
• Numaflow-python: 0.10.1 (also 0.12.1)

---

Message from the maintainers:

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

[vigith]

Do you mind sharing the pipeline spec? I would like to know what windowing you are using.

Notice logged datums coming into reduce UDF have a set _event_time but _watermark is effectively set to -1

This is the correct behavior if watermark is not moving. The question should be why is watermark not moving.

[niall-nai]

Hi @vigith thanks for the reply

Ideally I'd like to use a session window but I assume that's not supported (no note here: https://numaflow.numaproj.io/user-guide/user-defined-functions/reduce/windowing/session/ ). I've tried:

• Session with timeout 3m
• Sliding length 60s, slide 10s
• Fixed with length 60s

All with streaming and keyed set to true

The current instance I've been testing:

- name: [...]
    partitions: 1
    limits:
      readBatchSize: 1
    udf:
      container:
        args:
        - python
        - src/[...]
        image: localhost:5000/[...]
        imagePullPolicy: Always
      groupBy:
        window:
          sliding:
            length: 60s
            slide: 10s
            streaming: true
        keyed: true
        storage:
          persistentVolumeClaim:
            volumeSize: 1Gi
            accessMode: ReadWriteOncePod

and with a simple class inheriting from ReduceStreamer (code redacted, debug statements added):

async def handler(
        self,
        keys: list[str],
        datums: AsyncIterable[Datum],
        output: NonBlockingIterator,
        md: Metadata
    ):
        print(md.interval_window.start)
        async for datum in datums:
            print(datum)
            [...]

        print("loop finished")
        for out in final_out:
             await output.put(Message(str.encode(out), keys=[keys], tags=['...']))

I'm seeing the first two print statements (interval start, datum), but never the "loop finished" statement - this is what's led me to conclude there seems to be a problem with the windowing itself, as I understand that final statement is where the accumulated output should be returning

Please do let me know if any of those are flawed assumptions; this is the best I could figure out from examples and documentation

[vigith]

@niall-nai could you please try out the latest 1.7.3 release?

[niall-nai]

@BulkBeing thanks for the quick work on this! Glad you could reproduce and solve

@vigith I can try upgrading everything yes

[niall-nai]

@vigith can confirm this happens @ 0.12.1 and 1.7.3 (updated original post)

[vigith]

We will try to reproduce this on our end /cc @BulkBeing

[BulkBeing]

@niall-nai The fix in my PR is not yet released (only in main branch). I've pushed the code I used here https://github.com/BulkBeing/reproduce-issues/tree/main/numa-python-reduce-watermark
Above code adds pynumaflow dependency as a git url .

In your setup:

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

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

NOTE: The package is in a sub-directory.

If possible, can you reproduce the issue with the setup/code in the repo I linked and create a PR to that if any modifications are needed.

[niall-nai]

@BulkBeing so I have good news and bad news

The good news is that the reduce streamer is reaching the output stage with watermarks! Fantastically quick turnaround, thank you!

The bad news is that I'm getting the following:

TypeError: bad argument type for built-in operation
2026-03-11 15:25:04 CRITICAL Reduce Streaming Error: TypeError('bad argument type for built-in operation')
Traceback (most recent call last):
  File "/opt/app-root/lib64/python3.12/site-packages/pynumaflow/reducestreamer/servicer/task_manager.py", line 256, in process_input_stream
    await con_future
  File "/opt/app-root/lib64/python3.12/site-packages/pynumaflow/reducestreamer/servicer/task_manager.py", line 288, in write_to_global_queue
    res = reduce_pb2.ReduceResponse.Result(keys=msg.keys, value=msg.value, tags=msg.tags)

When calling:

for out in output:
            msg_out = json.dumps(out, indent=4, default=str)
            await output.put(Message(str(msg_out).encode(), keys=[keys], tags=['...']))

out in this case is a fairly straightforward object which converts into a string easily with json.dumps. Replacing with a debug string also yields the same response. Tried without tags as in your example above

Also just for my awareness- is it expected that the reducer container should "complete" in k8s? This is just for my understanding so I can tweak the pipeline correctly

[BulkBeing]

I was able to reproduce this stacktrace. The problem is in:

output.put(Message(str(msg_out).encode(), keys=[keys], tags=['...']))

The type of the value passed to the keys parameter becomes list[list[str]] as the keys is already a list[str] https://github.com/BulkBeing/reproduce-issues/blob/cdcf7ed0d85995ac64b10e1a0199f91c5925e6fa/numa-python-reduce-watermark/main.py#L40
The TypeError: bad argument type for built-in operation is a generic protobuf error if any of the passed argument has wrong type.

I will create a PR for better error messages.

Also just for my awareness- is it expected that the reducer container should "complete" in k8s? This is just for my understanding so I can tweak the pipeline correctly

No, it's not expected — the reducer should remain running. The container completing/terminating is a side effect of the unhandled TypeError crashing the process.