Vector not buffering in case of message_timeout_ms=0

[ShahroZafar]

We have a setup where kubernetes_logs source is used to read and push logs to kafka via kafka sink. In case of outage of kafka, with memory buffer in place. I see that queued messages keeps on increasing instead of storing them to buffer. This leads to continuous memory increase and OOMKill

Configs are like this

    acknowledgements:
      enabled: true
    api:
      address: 0.0.0.0:8686
      enabled: true
      playground: false
    data_dir: /vector-data-dir
    expire_metrics_secs: 300
    sinks:
      kafka:
        batch:
          max_bytes: 1000000
          max_events: 10000
          timeout_secs: 2
        bootstrap_servers: kafka:9092
        buffer:
          max_events: 4000
          type: memory
          when_full: block
        compression: zstd
        encoding:
          codec: json
        inputs:
        - kubernetes_logs
        librdkafka_options:
          client.id: vector
          request.required.acks: "1"
        message_timeout_ms: 0
        topic: ingress
        type: kafka
      prometheus_exporter:
        address: 0.0.0.0:9090
        buffer:
          max_events: 500
          type: memory
          when_full: block
        flush_period_secs: 60
        inputs:
        - internal_metrics
        type: prometheus_exporter
    sources:
      internal_metrics:
        type: internal_metrics
      kubernetes_logs:
        glob_minimum_cooldown_ms: 5000
        include_paths_glob_patterns:
        - /var/log/pods/**/*
        ingestion_timestamp_field: ingest_timestamp
        type: kubernetes_logs

[jszwedko]

Hmm, there may be internal buffering in the rdkafka library that is coming into play here too. You could try tweaking the rdkafka options. To confirm, do you believe the memory is being buffered in the sink? Or the source?

[ShahroZafar]

To me it seems like accumulation in the sink. Though I see that when buffer gets full, memory used by kubernetes_logs source increases as well.
However its weird that the message doesn't get stored in the buffer. It negates the concept of having the buffer.

[ShahroZafar]

By the way, is there any way to retry the failed message at the vector sink level?

[jszwedko]

Thanks for sharing the additional details. My best guess is that the events are being buffered internally in the librdkafka library. I'd suggest looking at the options mentioned on https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md and tweaking ones that seem relevant.

[jszwedko]

Messages that failed to be forwarded to Kafka should be retried.

[ShahroZafar]

From what I understand, if we set message_timeout_ms to finite value like 5 mins. When this 5 mins is up, vector should enqueue the message in the buffer which is not doing. Can you confirm this theory?

[jszwedko]

That option ends up setting message.timeout.ms which is documented on https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md