Channel and window close issue with Reduce Vertex/UDF

[PRATHAM2002-DS]

My usecase Description:

I am using Reduce with fixed window and keyed set to false with source and sink as Kafka. The vertex needs to accumulate data in given time window which is set to 60s and send it to another external service for performing operations on those events. (All config values mentioned are used for testing/poc). i will add both UDF and Pipeline code below.

Issue Description:

When the first event hits Reduce function the event stays in for loop that works on datum data stream, and the for loop does not terminate and process data even after 60 s are over. The frequency of events coming in my usecase is not fixed and continuous i.e 2 events may come in first second and Third can come after 2 minutes for example. The event stay in loop or channel until the event comes just before window closes. I Also tried same UDF with keyed set to True but getting similar behaviour.

Expected Behaviour:

I expected that Reduce UDF will accumulate what ever events come in 60 s and after window length is completed it will close channel and terminate loop to process these accumulated events, even if there is no continuous event hitting Reduce. So event after first few seconds if no event came I expected it to terminate loop/ close channel and process event accumulated in first few seconds.

Actual Behaviour:

The events coming in first few seconds stay in loop and loop does not get terminated/channel does not close even after 60s is completed. Suddenly if some another event comes just before this window length is completed the channel closes and process all previous windows accumulated messages.

Pipeline Code:

apiVersion: numaflow.numaproj.io/v1alpha1
metadata:
  name: nf-business-pipeline
  labels:
    numaflow.numaproj.io/component: pipeline
    numaflow.numaproj.io/name: nf-business-pipeline
    numaflow.numaproj.io/part-of: numaflow
    numaflow.numaproj.io/created-by: controller-manager
  annotations:
    numaflow.numaproj.io/description: A numaflow meta pipeline which will handle for business execution
    numaflow.numaproj.io/owner: #owner
    numaflow.numaproj.io/pipeline-name: nf-business-pipeline
    numaflow.numaproj.io/test: "true"
spec:
  templates:
    daemon:
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      imagePullSecrets:
        - name: github-docker-registry
      tolerations:
      - key: "kubernetes.azure.com/scalesetpriority"
        operator: "Equal"
        value: "spot"
        effect: "NoSchedule"
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot          
    job:
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      imagePullSecrets:
        - name: github-docker-registry
      tolerations:
      - key: "kubernetes.azure.com/scalesetpriority"
        operator: "Equal"
        value: "spot"
        effect: "NoSchedule"
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot
    # If using custom service accounts, set vertex to null
    vertex: null
    sideInputsManager:
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      imagePullSecrets:
        - name: github-docker-registry
      tolerations:
      - key: "kubernetes.azure.com/scalesetpriority"
        operator: "Equal"
        value: "spot"
        effect: "NoSchedule"
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot     
  vertices:
    - name: bl-cdc-source
      source:
        kafka:
          brokers:
            - # kafka broker
          topic: BUSINESS
          consumerGroup: nf-pipeline-bl-execution
          tls: null
          sasl: null
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      tolerations:
        - key: kubernetes.azure.com/scalesetpriority
          operator: Equal
          value: spot
          effect: NoSchedule
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot            
      imagePullSecrets:
        - name: github-docker-registry
      scale:
        min: 1
        max: 1
    - name: bl-cdc-processor
      udf:
        container:
          image: # image of my go script UDF
          command:
            - ./main
            - execute
            - business-runner
          resources: {}
        groupBy: 
          window:
            fixed:
              length: 60s
          keyed: true
          storage:
            emptyDir: {}
        builtin: null
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      tolerations:
        - key: kubernetes.azure.com/scalesetpriority
          operator: Equal
          value: spot
          effect: NoSchedule
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot            
      imagePullSecrets:
        - name: github-docker-registry
      serviceAccountName: numaflow-sa-default-ns
      scale:
        min: 1
        max: 1
    - name: bl-dlq-sink
      metadata:
        annotations:
          cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      imagePullSecrets:
        - name: github-docker-registry
      tolerations:
      - key: "kubernetes.azure.com/scalesetpriority"
        operator: "Equal"
        value: "spot"
        effect: "NoSchedule"
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 1
            preference:
              matchExpressions:
              - key: eks.amazonaws.com/capacityType   #AWS
                operator: In
                values:
                - SPOT
          - weight: 1
            preference:
              matchExpressions:
              - key: lifecycle     #Azure
                operator: In
                values:
                - spot
          - weight: 1
            preference:
              matchExpressions:
              - key: cloud.google.com/gke-provisioning  #GCP
                operator: In
                values:
                - spot
      sink:
        kafka:
          brokers:
            -  # kafka broker
          topic: "BUSINESS_DLQ"
      scale:
        disabled: false
        min: 1
        max: 1

        
  edges:
    - from: bl-cdc-source
      to: bl-cdc-processor
      conditions: null
    - from: bl-cdc-processor
      to: bl-dlq-sink
      conditions:
        tags:
          values:
            - dlq   

Script Code:

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/marketplace-go/common/atlas"
	"github.com/marketplace-go/common/cache/redis"
	es "github.com/marketplace-go/common/elasticsearch"
	"github.com/marketplace-go/common/heracles"
	gatekeeper "github.com/marketplace-go/common/locks"
	"github.com/marketplace-go/constant"
	"github.com/marketplace-go/scripts"
	"github.com/numaproj/numaflow-go/pkg/reducer"
	"github.com/rs/zerolog/log"
)


func (f *SumReducerCreator) Execute(args []string) {
	if err := reducer.NewServer(f).Start(context.Background()); err != nil {
		log.Err(err).Msg("Failed to start the reducer server")
		os.Exit(1)
	}
}

// SumReducerCreator implements the reducer.ReducerCreator interface which creates a reducer
type SumReducerCreator struct {
}

func (s *SumReducerCreator) Create() reducer.Reducer {
	return &Sum{sum: []string{}}
}

// Sum is a reducer that sum up the values for the given keys
type Sum struct {
	sum []string
}

// Reduce function processes multiple events within a time window
func (f *Sum) Reduce(ctx context.Context, keys []string, datums <-chan reducer.Datum, md reducer.Metadata) reducer.Messages {
	fmt.Println("BL: Reduce called")

	for datum := range datums { // Process each datum until the channel is closed
		// Retrieve the value from the current datum
		rawData := datum.Value()
		eventKey := string(rawData)

		// Append the data to the accumulated result
		f.sum = append(f.sum, eventKey)

		fmt.Println("BL: Event Key: " + eventKey)
	}

	// Print all accumulated data once the channel is fully consumed
	fmt.Println("BL: Final accumulated data: " + fmt.Sprintf("%v", f.sum))

	return nil // Exit the function after all datums are processed
}

func init() {
	scripts.Register(constant.BusinessLineageRunner, &SumReducerCreator{})
}

LOGS:

Logs showing inconsistency when keyed set to false:

BL: Reduce called
BL: Event Key: {"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Final accumulated data: [{"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}]
BL: Final accumulated data: [{"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}]

Log showing inconsistency when keyed set to true:

BL: Reduce called
BL: Event Key: {"assetGuid":"asset1","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"edgeLabel1","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"edgeLabel2","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Reduce called
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"edgeLabel3","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Event Key: {"assetGuid":"cc878da0-e9af-48f6-a52e-6d2f2d2b0a22","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}
BL: Final accumulated data: [{"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"e3afa960-1b2b-4cdb-8fd7-84a3ef620189","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","edgeLabel":"","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"} {"assetGuid":"a9df1b19-0e0c-4ed6-8d94-6c258f265333","operation":"ADD","productGuid":"b72034c4-be31-44cd-9fca-2ade5d5295fb"}]

Note All logs after "Reduce called" indicate different window and finally due to some event coming at Reduce UDF just before some window length completes, all grouped messages from different windows got processed.

Question:

-> Is Reduce correct for my usecase ?
-> When I used storage config it stored and retried the unprocessed event but never removed those event even after successful retry, do we need to handle this explicitly? Config shown below:

storage:
            persistentVolumeClaim:
              volumeSize: 2Gi
              accessMode: ReadWriteOnce

[vigith]

are you incrementing the event-time in the events at the source? the 60s is based on the how the time progresses at source.

[PRATHAM2002-DS]

Another way I tried to do chunking was using Map and in-memory batching but on pod restart I was losing data. Hence I was trying a poc with reduce.

[PRATHAM2002-DS]

The behaviour I have mentioned above (in original question) with reduce, is it an expected behaviour?

[vigith]

return nil // Exit the function after all datums are processed .. you should return the output of Reduce function so it can be forwarded to the next vertex.

[PRATHAM2002-DS]

I will surely try this, but if you see logs the evnts coming in first few seconds stay in the for-loop as channel does not get closed and do not come to this return statement even after window length is completed. These events only exit this loop when another event hits this function just before some other windows length is completed.

[vigith]

These events only exit this loop when another event hits this function just before some other windows length is completed.

That is due to idling, I have pasted the link in the discussion

[vigith]

https://numaflow.numaproj.io/user-guide/user-defined-functions/reduce/windowing/windowing/

[vigith]

if you set key to true, you will have to set the keys in the pervious map vertex.

[vigith]

So event after first few seconds if no event came I expected it to terminate loop/ close channel and process event accumulated in first few seconds.

you will have to turn on idle detection if your source is not emitting data.

[PRATHAM2002-DS]

Can you help me with following question @vigith

When I used storage config it stored and retried the unprocessed event but never removed those event even after successful retry, do we need to handle this explicitly? Config shown below:

storage:
            persistentVolumeClaim:
              volumeSize: 2Gi
              accessMode: ReadWriteOnce
  

[PRATHAM2002-DS]

After every new event or restart the store put some stored previous events in channel to process again.

[yhl25]

After the window is successfully closed and the responses are forwarded to the next vertex, the events of that window will be cleared from the storage. Incase of restarts we replay the events that are persisted in the storage before reading the new set of events to make sure there is no data loss, after the window is closed those events will be cleared from the store.

[PRATHAM2002-DS]

@yhl25 @vigith if I use keyed events , set keyed: true for this reduce vertex and partitions:2 value to 2. Does this guarantee ordering? Will all events with same key be consumed by one partition?

[vigith]

Does this guarantee ordering?

There will be temporal ordering, the key for same window will be send to the same reduce function.

Will all events with same key be consumed by one partition?

Yes, same key will end up in the same reduce function for that window.

[PRATHAM2002-DS]

@yhl25 @vigith suppose I set window length as 60s and the processing for events in first window has not yet completed and events from second window (say 60-120s) hit reduce function, does this affect events from first window? Will I loose the unprocessed events from first window or any other side effect that I should take care of?

[yhl25]

It should not affect, windows are unique per reducer. Messages of one window will always be forwarded to one instance of the reducer. They never overlap, and one window should not affect another.

[PRATHAM2002-DS]

@yhl25 in this case where the second window will not affect and not wait till all events from first window are processed completely, the order of executing events wont be maintained right?

[vigith]

second window is completely independent of the first window. since there is a temporal causality within the windows, second cannot close before first, so ordering is not lost.