Kafka source low throughput

[Arty-Maly]

Hello, I have created a simple Vector pipeline that looks like this: Kafka -> Remap -> Kafka.

I am noticing a low amount of messages coming from the Kafka source no matter what configuration I set or the amount of partitions I have. The amount of events in bytes emitted looks to be ~1MB per second.

The utilization of my remap is low as well. Around 0.1. The rate is the same if I have lag or not (Kafka Broker has more messages in its queue)

Now my Pipeline can keep up with the Kafka source queue if there are no hiccups in the system. If I restart Vector and let lag build up, Vector will not be able to clear it.

I have also tried to verify the bottleneck is in the Kafka Source and just tried a pipeline in the form of:
1 )Kafka -> Black Hole
2) Kafka -> Remap -> Black Hole

Both of those designs have the same symptoms. Kafka source returning emitting a low number of events to the next step.

Acknowledgements are turned on and setting or removing the encoding: json in the source doesnt make a difference.

A simple example of my pipeline would be:

sources:
      kafka:
         type: kafka
         topics: "my-topic"
         bootstrap_servers: "my.server:port"
         group_id: "my-group"
     transforms:
        extract:
            type: remap
            inputs:
               - kafka
            source: |
                . = parse_json!(.)
        sinks:
             emit_to_kafka:
             inputs: 
                 - extract
             ....

I have tried setting the kafka options in the source: fetch.max.bytes, max.poll.records and broker options as well.
The only setting to have a difference was fetch.wait.max.ms but that just increased/decreased overall request rate with the amount of data returned over a period staying the same.

Is there anything that I am missing or need to verify as to why my kafka source is not saturating my pipeline?

[pront]

Hi @Arty-Maly, using the internal metrics source here would help to inspect the Kafka source metrics. There are many interesting metrics here such as kafka_consumer_lag, component_received* and component_sent_* that might explain what's slow in this pipeline.

[Arty-Maly]

Yes I narrowed it down to the Kafka source its not receiving enough messages from the kafka queue. I am wondering if its only a single thread doing the work for the source?

There are no discarded messages, the librdkafka queue is 0.
component_received and component_sent_ line up.
When I have more messages in the queue (lag) the source does not pick up more messages. So My pipeline is in a stable lagging state.

I also compiled the project myself, with release profile and the features I am intereseted in : transforms-remap, sources,sinks-blackhole,sinks-kafka

[Arty-Maly]

The same pattern occurs if I have 1 partion or 30

[Arty-Maly]

I receive about 8 events for the kafka source and send the same amount, the graph is identical. My next step then receives those events and sends them to the sink. But the utilization on that step is low, indicating it is waiting for input data.

[pront]

Hi @Arty-Maly, I didn't have time to dive into this. One guess, did you overwrite any librdkafka_options? Can you test with another topic to see if the rates are the same?

[alexeynl]

@Arty-Maly
Having the similar issue. Wanted to try max.poll.records but it is not part of librdkafka_options.

2025-02-28T14:41:29.313730Z ERROR vector::topology::builder: Configuration error. error=Source "kafka-source": Could not create Kafka consumer: Client config error: No such configuration property: "max.poll.records" max.poll.records 10000

How did you get this works?