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

[whynowy]

There is an example in https://numaflow.numaproj.io/user-guide/reference/pipeline-tuning/ for configuration in vertex level.

  vertices:
    - name: cat
      udf:
        builtin:
          name: cat
      limits:
        readBatchSize: 200 # It overrides the default limit "100"

[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