Kafka sink batch messages

[luk-ada]

Hello

I'm using Kubernetes logs as a source and Kafka sink as destination to send logs to the Azure Event Hub. After reviewing number of messages in the Event Hub and in log search engine it looks like 1 Event == 1 Message in the Event hub. Recently we had a problem with reading lot off messages produced by one AKS cluster which is based in the different Azure Region (EH also in different region). We are reading messages using Vectors. No problems in the same region, 1 Vector is able to work on similar load.
I would like to batch messages before sent to the Event Hub to have more logs in one message.
I tried every option from batch. Still it seams like single messages per log.

Vector 0.24.1

Sink config:

sinks:
  event_hub:
    type: kafka
    inputs: ["kubernetes_logs"]
    bootstrap_servers: "${NAMESPACE}.servicebus.windows.net:9093"
    topic: "${EVENT_HUB}"
    compression: "none"
    encoding:
      codec: "json"
    librdkafka_options:
      "security.protocol": sasl_ssl
      "sasl.mechanism": "PLAIN"
      "sasl.username": "$$ConnectionString"
      "sasl.password": "${PASSWORD}"
      # https://learn.microsoft.com/en-us/azure/event-hubs/apache-kafka-configurations#librdkafka-configuration-properties
      "socket.keepalive.enable": "true"
      "metadata.max.age.ms": "180000"
      "retries": "2"
      "request.timeout.ms": "60000"
      "partitioner": "consistent_random"

[tobz]

The kafka sink is explicitly designed to maintain a "one event equals one Kafka message" relationship. The batching configuration affects how the underlying Kafka client library, librdkafka, will do send operation batching. This can make sending messages to Kafka brokers more efficient, but it doesn't change the "one event equalis one Kafka message" behavior.

One approach you could try is to aggregate your events in a reduce transform before they reach the kafka sink. This would involve something like:

transforms:
  wrap_event:
    type: remap
    inputs:
      - name_of_component
    source: |-
      . = { "events": . }
  aggregated_events:
    type: reduce
    inputs:
      - wrap_event
    merge_strategies:
      events: array
sinks:
  kafka_output:
    type: kafka
    inputs:
      - aggregated_events

What this is doing is:

• takes the original event and wraps it in an object, where the original event is now set on the events field (so an original event that looks like { "foo": "bar" } becomes { "events": { "foo": "bar" } })
• aggregate those events by taking the value of the events field and pushing it into an array (so the event emitted by reduce looks like { "events": [{ "foo": "bar" }, { "foo": "bar" }, ...] })
• feed those reduced events into your kafka sink

We're sort of abusing reduce here to get the intended behavior. This means that, by default, it will flush itself every 1 second with all of the events it has reduced so far: it doesn't have a way to specify emitting an event after it's reduced a specific number of input events, so it's not quite the same as the sink batching setting.

[luk-ada]

@tobz thank you for your answer. Unfortunately it's not working very well as we are not able to specify how many messages can be wrapped together. After I enable those transformations number off messages dropped immediately to 2-3 per minute (from 10k) and only less than 0,01% logs were sent. I will play with expire_after_ms and flush_period_ms as lowering those values results to bring back more logs like 10% of previous load.
I'm open for other other suggestions as it's very hard to say if we are not dropping any messages :(

[tobz]

It might be worth using the console sink while testing out that solution, to write the merged events to stdout and make sure they're in the right format, and arriving at the expected rate, etc.

[luk-ada]

I think that solution for this problem will be implementation of reduce and max_events feature from #14817 (when will be available).