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
• Numaflow-python: 0.10.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