How to improve the consumption capacity

[loveyang2012]

I am using Vector to collect Kafka messages, with two sinks: one is an HTTP-based OpenObserve interface, and the other is a single-node Kafka. The current issue is that the consumption capacity is insufficient, with Vector pulling around 15,000 messages per second, totaling about 5MB in size. In the case of multiple sinks, could one of the sinks impact the overall consumption speed? If it's a configuration issue with Vector, what configurations can be optimized to improve the consumption capacity?

vector version : 0.38

[jszwedko]

Can you share your configuration?

In the case of multiple sinks, could one of the sinks impact the overall consumption speed?

Yes, see https://vector.dev/docs/about/under-the-hood/architecture/buffering-model/#backpressure-and-the-need-to-buffer

[loveyang2012]

here

sources:
  a_from_kafka:
    type: kafka
    bootstrap_servers: 10.230.13.18:9092,10.230.13.17:9092,10.230.13.15:9092
    group_id: vector-prod
    topics:
     - push-msgcenter-znx
transforms:
  b_get_message:
   type: remap
   inputs: 
     - "a_from_kafka"
   source: |
     . = parse_json!(.message)
sinks:      
  to_znx_kafka:
    type: kafka
    inputs:
      - "b_get_message"
    bootstrap_servers: sz-core-l-zgkhd-c3-1-20:9092
    topic: feishu-message-topic
    encoding:
      codec: "json"
  openobserve:
    type: "http"
    inputs:
     - "b_get_message"
    uri: "http://10.230.13.13:8080/cxlog/api/default/push-msgcenter-znx/_json"
    method: "post"
    compression: "gzip"
    auth:
      strategy: "basic"
      user: "xx"
      password: "xx"
    encoding:
      codec: "json"
      timestamp_format: "rfc3339"
    healthcheck:
      enabled: false

[jszwedko]

Thanks! I'd suggest:

• Identifying which sink is the bottleneck. You should be able to look at the utilization internal metric to determine that (the one with the higher value is likely the bottleneck)
• If it is the kafka sink you could play around with the librdkafka options: https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md
• If it is the elasticsearch sink you could try setting request.concurrency to a static integer (it defaults to "adaptive" where it tries to find the optimal concurrency)