Kafka

Synopsis

Creates a collector that connects to Kafka brokers and consumes messages from specified topics. Supports authentication, TLS encryption, and multiple workers.

For details, see Appendix.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: kafka
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    username: <string>
    password: <string>
    algorithm: <string>
    kerberos_keytab: <string>
    kerberos_principal: <string>
    kerberos_realm: <string>
    kerberos_kdc: <string>
    kerberos_config: <string>
    kerberos_service_name: <string>
    group: <string>
    topic: <string>
    balancer: <string>
    reuse: <boolean>
    workers: <numeric>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description
type	Y	-	Must be kafka
tags	N	-	Optional tags
pipelines	N	-	Optional pre-processor pipelines
status	N	true	Enable/disable the device

Connection

Field	Required	Default	Description
address	N	"0.0.0.0"	Kafka broker address
port	Y	-	Kafka broker port
username	N	-	SASL username
password	N	-	SASL password
algorithm	N	-	SASL mechanism: plain, scram-sha-256, scram-sha-512, or gssapi. Leave empty or set to none to disable authentication
kerberos_keytab	N*	-	Keytab file path, or base64-encoded keytab content. Resolvable via ${ENV_VAR} or $secret{...}
kerberos_principal	N	username	Client principal (user or user@REALM). Falls back to username when omitted
kerberos_realm	N	-	Kerberos realm. Overrides the realm derived from the principal or krb5.conf
kerberos_kdc	N*	-	Comma-separated KDC addresses (host[:port], default port 88)
kerberos_config	N*	-	krb5.conf file path, or inline krb5.conf content. Generated from kerberos_realm and kerberos_kdc when omitted
kerberos_service_name	N	"kafka"	Kafka broker service principal name (SPN)
group	N	"vmetric"	Consumer group ID
topic	Y	-	Topic to consume from
balancer	N	"roundrobin"	Partition assignment strategy: cooperativesticky, range, roundrobin or sticky. Matched case-insensitively but without hyphens — an unrecognized value is not rejected, it silently becomes roundrobin

* = Applies only when algorithm is gssapi. kerberos_keytab is required, along with either kerberos_kdc or kerberos_config.

TLS

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption
tls.cert_name	Y*	cert.pem	Client certificate for mutual TLS
tls.key_name	Y*	key.pem	Client private key for mutual TLS
tls.passphrase	N	-	Passphrase for an encrypted private key
tls.ca_name	N	-	CA bundle used to verify the broker certificate. When unset, the host trust store is used.
tls.min_tls_version	N	tls1.2	Minimum accepted TLS version (tls1.0, tls1.1, tls1.2, tls1.3)
tls.max_tls_version	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
tls.insecure_skip_verify	N	false	Skip broker certificate verification. Use only for testing.

* = Conditionally required when tls.status is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Performance

Field	Required	Default	Description
reuse	N	true	Enable multi-worker mode
workers	N	4	Number of worker processes when reuse enabled (capped at the number of available CPU cores)
max_buffer_size	N	128MB	Ingest buffer admission cap. Also sets this listener's eager heap reservation

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, which commits 384 MiB of heap per listener worker before any data arrives — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

Examples

Basic

Creating a simple Kafka consumer...

- id: 1 name: basic_kafka type: kafka properties: address: "kafka.example.com" port: 9092 topic: "logs"

Secure

Connecting with SASL authentication and TLS encryption...

- id: 2 name: secure_kafka type: kafka properties: address: "kafka.example.com" port: 9093 username: "consumer" password: "secret" algorithm: "scram-sha-256" topic: "secure-logs" tls: status: true cert_name: "kafka.crt" key_name: "kafka.key"

Kerberos

Authenticating with Kerberos (GSSAPI) using a keytab and KDC...

- id: 6 name: kerberos_kafka type: kafka properties: address: "kafka.example.com" port: 9093 algorithm: "gssapi" kerberos_keytab: "$secret{id=12}" kerberos_principal: "vmetric@EXAMPLE.COM" kerberos_kdc: "kdc01.example.com,kdc02.example.com:88" kerberos_service_name: "kafka" topic: "secured-logs" tls: status: true cert_name: "kafka.crt" key_name: "kafka.key"

High-Volume

Optimizing for throughput with multi-worker mode...

- id: 3 name: performant_kafka type: kafka properties: address: "kafka.example.com" port: 9092 topic: "high-volume-logs" group: "high-perf-group" reuse: true workers: 4

Consumer Groups

Configuring consumer group behavior...

- id: 4 name: group_kafka type: kafka properties: address: "kafka.example.com" port: 9092 topic: "shared-logs" group: "processing-group" balancer: "roundrobin" reuse: true workers: 2

warning

Consumers in the same group must use compatible configuration settings.

Pipelines

Applying custom processing to messages...

- id: 5 name: pipeline_kafka type: kafka pipelines: - json_parser - field_extractor properties: address: "kafka.example.com" port: 9092 topic: "raw-logs" group: "processing-group"

note

Pipelines are processed sequentially, and can modify or drop messages before ingestion.