Memory usage of numa container in reduce UDF pod increases monotonically (former: numa container in reduce UDF pod got down by "No space left on device")

[tmenjo]

Describe the bug

I have a pipeline stream-joining video frames and inference results as follows. The reduce UDF vertex accumulator(stream-join) does that join.

[sourcer(input)] ---------------+
    | Original-sized frames     | Original-sized frames
    v                           |
[mapstreamer(resize)]           |
    | Scaled-down frames        |
    v                           |
[mapstreamer(inference)]        |
    | Inference results         |
    v                           |
[accumulator(stream-join)] <----+
    | 
    | Messages joined on frame index
    v
[sinker(output)]

Running that pipeline, I found that the numa container in the reduce UDF vertex pod got down. Its logs said that Failed to write message to WAL and No space left on device. I also found that memory usage of the numa container increased almost monotonically.

To Reproduce

1. Run kubectl apply -f pipeline1.yaml
2. Wait for a while

pipeline1.yaml

The part of the reduce vertex in pipeline1.yaml:

    - name: stream-join
      partitions: 1
      scale:
        min: 1
        max: 1
      udf:
        container:
          image: 192.168.21.17:30003/dci-community-general/dci-poc:stable
          imagePullPolicy: IfNotPresent
          env:
            - name: SCRIPT
              value: reduce
          envFrom:
            - configMapRef:
                name: app-env-cm
          volumeMounts:
            - mountPath: /var/tmp/logs/numaflow-dra/dci_poc
              name: log-volume
        groupBy:
          window:
            accumulator:
              timeout: 1s
          keyed: false
          storage:
            persistentVolumeClaim:
              storageClassName: local-storage
              volumeSize: 3Gi
              accessMode: ReadWriteOnce
      limits:
        readBatchSize: 1
      volumes:
        - name: log-volume
          hostPath:
            path: /var/tmp/logs/numaflow-dra/dci_poc
            type: DirectoryOrCreate

Expected behavior

The numa container continues running.

Logs and screenshots

numa.log (numa container's full logs)

The tail of the logs:

{"timestamp":"2026-02-25T03:18:06.062212Z","level":"INFO","message":"Rotating WAL segment file","current_size":"10335304","file_name":"\"/var/numaflow/pbq/data_437_1771989485410332.wal\"","target":"numaflow_core::reduce::wal::segment::append"}
{"timestamp":"2026-02-25T03:18:06.063321Z","level":"INFO","message":"rename successful","self.current_file_name":"\"/var/numaflow/pbq/data_437_1771989485410332.wal\"","to_file_name":"\"/var/numaflow/pbq/data_437_1771989485410332.wal.frozen\"","target":"numaflow_core::reduce::wal::segment::append"}
{"timestamp":"2026-02-25T03:18:06.190208Z","level":"INFO","message":"Processed messages per second","processed":"31","target":"numaflow_core::tracker"}
{"timestamp":"2026-02-25T03:18:06.416833Z","level":"ERROR","message":"Failed to write message to WAL","e":"Io(Os { code: 28, kind: StorageFull, message: \"No space left on device\" })","target":"numaflow_core::reduce::wal::segment::append"}
{"timestamp":"2026-02-25T03:18:06.416979Z","level":"INFO","message":"Nak received for offset","params.offset":"Int(IntOffset { offset: 8855, partition_idx: 0 })","target":"numaflow_core::pipeline::isb::reader"}
{"timestamp":"2026-02-25T03:18:06.417037Z","level":"INFO","message":"Nak sent for offset","offset":"Int(IntOffset { offset: 8855, partition_idx: 0 })","target":"numaflow_core::pipeline::isb::reader"}
{"timestamp":"2026-02-25T03:18:06.418078Z","level":"ERROR","message":"/app/numaflow-core/src/reduce/pbq.rs:146:18: Receiver dropped: SendError { .. }","target":"numaflow::setup_tracing"}
{"timestamp":"2026-02-25T03:18:06.418486Z","level":"INFO","message":"Unaligned Reduce component is shutting down, waiting for active reduce tasks to complete","target":"numaflow_core::reduce::reducer::unaligned::reducer"}
{"timestamp":"2026-02-25T03:18:06.418515Z","level":"INFO","message":"Waiting for 1 active reduce tasks to complete","target":"numaflow_core::reduce::reducer::unaligned::reducer"}
{"timestamp":"2026-02-25T03:18:06.418677Z","level":"ERROR","message":"Error while joining PBQ reader and reducer","e":"JoinError::Panic(Id(35), \"Receiver dropped: SendError { .. }\", ...)","target":"numaflow_core::pipeline::forwarder::reduce_forwarder"}
{"timestamp":"2026-02-25T03:18:06.418759Z","level":"INFO","message":"Stopped the Lag-Reader Expose tasks","target":"numaflow_core::metrics"}
{"timestamp":"2026-02-25T03:18:06.418939Z","level":"ERROR","message":"Error running pipeline","e":"Forwarder(\"Error while joining PBQ reader and reducer: task 35 panicked with message \\\"Receiver dropped: SendError { .. }\\\"\")","target":"numaflow_core"}
{"timestamp":"2026-02-25T03:18:06.419441Z","level":"INFO","message":"Gracefully Exiting...","target":"numaflow_core"}
{"timestamp":"2026-02-25T03:18:06.419473Z","level":"INFO","message":"Exited.","target":"numaflow"}
{"timestamp":"2026-02-25T03:18:06.421082Z","level":"ERROR","message":"TrackerState dropped with 1 unacknowledged messages still tracked","total_entries":1,"target":"numaflow_core::tracker"}

Memory usage of the reduce pod:

Environment

• Kubernetes: [e.g. v1.33.2]
• Numaflow: [e.g. nightly-20260206]

[vigith]

Was the watermark progressing?

[vigith]

This usually happens when watermark does not progress causing windows to not getting closed. If the data to one source is rare, make sure idling is configured (e.g., https://github.com/numaproj/numaflow-python/blob/main/packages/pynumaflow/examples/accumulator/streamsorter/pipeline.yaml#L8-L18)

[tmenjo]

@vigith Thanks for your advice. Idling is not configured.

I have a question before trying. The data to my source is a streaming video in 15fps. It continues forever with no break ideally. In that case, will an acuumulator window close? If won't, do you think this issue will happen again? I'm concerned whether my usage of accumulator is proper or not.

[vigith]

The data to my source is a streaming video in 15fps. It continues forever with no break ideally. In that case, will an acuumulator window close?

Then it should close, does the UI show watermark progression?

If won't, do you think this issue will happen again? I'm concerned whether my usage of accumulator is proper or not.

Why are you using accumulator? Why not session window or aligned windows (fixed or sliding) ? Accumulator is used for solving a very low level ordering problem.

[tmenjo]

@vigith It has been a long time in this thread.

Watermark progression

Then it should close, does the UI show watermark progression?

Here is a screenshot. Watermark advences, as far as I can see. Watermark lag keeps 1 or 2 seconds.

• Kubernetes: v1.35.2
• Numaflow: nightly-20260521
• numaflow-python: v0.13.0

---

Accumulator or session window

Why are you using accumulator? Why not session window or aligned windows (fixed or sliding) ? Accumulator is used for solving a very low level ordering problem.

I'm embarrassed to say that I'm not confident whether session window or accumulator is more suitable for our application, numaflow-dra. Our whole pipeline is as shown in the following figure:

Our reduce vertex is the Union in the figure. Its code is here.

For now, the Union is a non-keyed accumulator. It takes two input streams. One is a video frame stream, and the other is an inference result stream on each video frame. Each datum (frame or result) in each stream also has a frame index. The Union accumulates two input streams and merges a frame and a result having the same frame index into one output datum for drawing bounding boxes. I think this usecase is a stream joining, described in the Accumulator page of the Numaflow documentation, so I have implemented the Union as an accumulator.

It would be very helpful if you could take a look into my Union code and give me some advice whether I should use session window or accumulator.

[vigith]

Just confirming, even if the watermark progresses you see that WAL is not getting cleaned up. Can you please share the accumulator logs?

Any reason why you are using non-keyed, can't you use keyed and use frame-index as the key?

Anyways, let's discuss this during tomorrow's numaflow bi-weekly call.

[tmenjo]

@vigith Sorry for missing the meeting but I took a day off on the last Friday...

Any reason why you are using non-keyed, can't you use keyed and use frame-index as the key?

Thank you for taking the time to help me. Here is what I did:

• My accumulator is now keyed and using a persistent volume.
• Each datum has [str(int(frame_index/900))] as keys in order to open a window every 60 seconds.
  • My source is a 15fps video. 900 frames for 60 seconds.
  • Please note that I had a performance issue when keys was simply [str(frame_index)].

Accumulator configuration:

        groupBy:
          window:
            accumulator:
              timeout: 5s
          keyed: true
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi

Memory usage (for 16 hours) is as follows. Compared with the previous non-keyed accumulator, memory usage of numa container (yellow) drastically felt down. However, it still did not reached a ceiling. Also, that of udf container (blue) tended to increase slightly.

Can you please share the accumulator logs?

Sorry but I forgot to log for that time. I'm working it. Please stay tuned.

[tmenjo]

I've had another run for 3 hours with debug logging.

Memory usage is as follows. The blue line is for the numa container. There are a few bumps at 00:24:00-00:25:00 and 02:34:00-02:38:00.

Debug logs:

• numa-20260604-001600.tar.gz
• numa-20260604-003000.tar.gz
• numa-20260604-010000.tar.gz
• numa-20260604-013000.tar.gz
• numa-20260604-020000.tar.gz
• numa-20260604-023000.tar.gz
• numa-20260604-030000.tar.gz

Reduce and debug configuration (see also #3453):

        groupBy:
          window:
            accumulator:
              timeout: 5s
          keyed: true
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi
      containerTemplate:
        env:
          - name: NUMAFLOW_DEBUG
            value: !!str "true"
          - name: RUST_LOG
            value: debug,h2::codec=info

Environment:

• Kubernetes: v1.35.2
• Numaflow: v1.8.0
• pynumaflow: v0.13.0

[yhl25]

Could you please provide a high-level overview of your accumulator implementation? Do you generate a response for each request individually, accumulate multiple requests and send a single response (1:1 vs. n:1), or drop some requests under certain conditions?