Rate Limiting Message Processing from Kafka Source: Best Practices for Consistent Processing

[4rpit]

I'm working with a processing pipeline where Kafka is the source of messages. Currently, the processing rate is unbounded, leading to inconsistent processing rates and potential downstream bottlenecks. what configuration I can add to the source for a better and bounded processing rate

[vigith]

To do throttling on a particular vertex you can do the following:

• reduce the readBatchSize of the vertex, set it to a number such that your downstream can handle concurrently (batchSize * number of pods will be the approximate concurrent requests on your downstream)
• reduce the number of max pods (scale.max) of the vertex so it won't autoscale to something your downstream cannot handle. this can be achieved via autoscaling

[4rpit]

hey folks back again, I did some experiments with this number and was not able to do throttling on the source vertex - even if I keep the readBatchSize either 1 or 100 it produces the same load downstream could the value for DefaultKafkaHandlerChannelSize of 100 is the reason it is keeping 100 messages from Kafka and not following the limits in the Vertex

[vigith]

could you please share your pipeline config?

[4rpit]

kind: Pipeline
apiVersion: numaflow.numaproj.io/v1alpha1
metadata:
  name: pipeline-name
  namespace: default
  labels:
  annotations:
  finalizers:
    - pipeline-controller
spec:
  vertices:
    - name: kafka-source
      source:
        kafka:
          brokers:
            - kafka-headless.kafka.svc.cluster.local:9092
          topic: CDC
          consumerGroup: ng-cg-1
      limits:
        readBatchSize: 1
      scale:
        min: 1
        max: 2
    - name: processor
      udf:
        container:
          image: image-id
    - name: post-processor
      udf:
        container:
          image: image-id
    - name: sink
      sink:
        log: {}
  edges:
    - from: kafka-source
      to: processor
      conditions: null
    - from: processor
      to: post-processor
      conditions: null
    - from: post-processor
      to: sink
      conditions: null

this is the pipeline (omit few details)

[vigith]

add limits to the UDF

    - name: processor
      udf:
        container:
          image: image-id
      limits:
        readBatchSize: 1

[4rpit]

kindly help me validate my understanding -> Source will poll a batch of messages but udf will only poll from the source based on the limit. or please point to any doc where I can read through it

[vigith]

Source writes to ISB and that data is fed in to the UDF. UDF does not poll from source.

[syayi]

@4rpit Thank you for exploring Numaflow. Apologies for asking this in the discussion thread, but could you share how you discovered the Numaflow project? Any specific use cases that you're exploring for? Helps us to learn what's resonating with the community :)

[4rpit]

hey @syay - in my organisation, we are using numaflow for different processing pipelines like notification, processing change events etc that's how I am aware of it

[syayi]

Thank you @4rpit ! Atlan?