Kafka

Synopsis

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

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>

Configuration

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 balancing strategy

* = 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*	-	TLS certificate file name
tls.key_name	Y*	-	TLS private key file name

* = Conditionally required when tls.status is true.

note

The TLS certificate and key files must be placed in the service root directory.

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)

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.