UDSink receives one unexpected message during shutdown

[KeranYang]

Describe the bug

I have a testing pipeline generator -> cat -> java udsink. I killed the pipeline using kubectl delete -f and at sink, I observed this message being received. id: cat-49-0-0, value: [], header: {}, eventtime: 2024-12-13T18:41:39.424879807Z, watermark: 0001-01-01T00Z.

We should never expose to UDFs messages that are not produced by previous vertices.

To Reproduce
Steps to reproduce the behavior:

Numaflow Version: 1.4.0+
Java SDK Version: 0.9.0

Use the following pipeline spec. run kubectl delete -f.

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: test-pipeline
spec:
  vertices:
    - name: in
      scale:
        min: 1
      source:
        generator:
          rpu: 1
          duration: 1s
    - name: cat
      scale:
        min: 1
      udf:
        builtin:
          name: cat
    - name: sink
      scale:
        min: 1
      limits:
        readBatchSize: 5
      sink:
        udsink:
          container:
            image: quay.io/numaio/numaflow-go/sink-log:stable
            imagePullPolicy: Always
  edges:
    - from: in
      to: cat
    - from: cat
      to: sink

See the last message being printed out:

User Defined Sink: {"Data":{"value":1734119381121003775},"Createdts":1734119381121003775}
User Defined Sink: {"Data":{"value":1734119382121008150},"Createdts":1734119382121008150}
User Defined Sink: {"Data":{"value":1734119383121005400},"Createdts":1734119383121005400}
User Defined Sink: {"Data":{"value":1734119384121005775},"Createdts":1734119384121005775}
User Defined Sink: {"Data":{"value":1734119385121005192},"Createdts":1734119385121005192}
User Defined Sink: 
2024/12/13 19:49:45 gracefully stopping grpc server
2024/12/13 19:49:45 error receiving from sink stream: context canceled
2024/12/13 19:49:45 error sending sink response: rpc error: code = Canceled desc = context canceled
2024/12/13 19:49:45 Stopping the SinkFn with err, context canceled

Message from the maintainers:

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

For quick help and support, join our slack channel.

[KeranYang]

Same behaviour observed with Go SDK - v0.9.0.

[vigith]

could be a regression due to streaming gRPC, do you see this with 1.3.3?

[KeranYang]

@vigith @whynowy I just tested, didn't see this with 1.3.3.

[KeranYang]

monovertex doesn't have such issue.

[whynowy]

@KeranYang - I believe this issue also happens in v1.3.3 (reproduced it). But it's not caused by WMB but the generator clean shutdown problem, it always generates lots of empty messages at the time being shut down, and one of them is propagated to the downstream (dedup).

There are some small things can be improved. #2288

[KeranYang]

@whynowy Thanks for looking into this issue! It explains why mvtx doesn't have this issue - I used an udsource to test.

Not sure why I didn't see it when I use https://github.com/numaproj/numaflow/blob/release-1.3/config/install.yaml to deploy numaflow. I assumed it's v1.3.3. Maybe it's transient.