Kafka

简介

service_kafka input插件实现了ServiceInputV1和ServiceInputV2接口，插件用于采集Kafka的消息。

事件能力

列含义见 概览 · 事件能力列说明。

Log	Metric	Span
✓	—	—

版本

Stable

版本说明

• 推荐版本：LoongCollector v3.0.5 及以上

配置参数

参数	类型	是否必选	说明
Type	string	是	插件类型，指定为service_kafka。
Format	string	否	自 1.6.0 起新增（同仓历史版本号，旧文档常写作 ilogtail），仅 ServiceInputV2 支持。v2 版本支持格式：raw、prometheus、otlp_metricv1、otlp_tracev1 说明：raw 格式以原始请求字节流传输数据，默认值：raw
Version	string	是	Kafka集群版本号。
Brokers	Array	是	Kafka服务器地址列表。
ConsumerGroup	string	是	Kafka消费组名称。
Topics	Array	是	待消费的Kafka订阅主题列表。
ClientID	string	是	消费Kafka的用户ID。
Offset	string	否	Kafka初始消费位移类型，可选值包括：oldest和newest。如果未添加该参数，则默认使用oldest，表示从最早可用的位移处开始消费。
MaxMessageLen（Deprecated）	Integer	否	Kafka 消息的最大允许长度，单位为字节，取值范围为：1～524288。若未添加该参数，则默认 524288（512KB）。自 1.6.0 起已不再使用此参数（旧文档常写作 ilogtail）。
SASLUsername	string	否	SASL用户名。
SASLPassword	string	否	SASL密码。
Assignor	string	否	消费组消费分区分配策略。可以设置选项：range, roundrobin, sticky，默认值：range
DisableUncompress	bool	否	自 1.6.0 起新增（旧文档常写作 ilogtail），禁用对请求数据的解压缩，默认 false。 目前仅针对 Raw Format 有效。 仅 v2 版本有效。
FieldsExtend	bool	否	是否支持非integer以外的数据类型(如String) 目前仅针对有 string、Bool 等额外类型的 influxdb Format 有效，仅v2版本有效

样例

采集服务器地址为172.xx.xx.48和172.xx.xx.34、主题为topicA和topicB的Kafka消息，并将采集结果输出至标准输出，其中Kafka集群的版本为2.1.1，消费组的名称为test-group，其余取默认值。

• 输入

{"payload":"foo"}

采集配置（v1）

enable: true
inputs:
  - Type: service_kafka
    Version: 2.1.1
    Brokers: 
        - 172.xx.xx.48
        - 172.xx.xx.34
    ConsumerGroup: test-group
    Topics:
        - topicA
        - topicB
    ClientID: sls
flushers:
  - Type: flusher_stdout
    OnlyStdout: true  

• 输出

{"payload":"foo"}

采集配置（v2）

v2 是自 1.6.0 起新增的实现（旧文档常写作 ilogtail），主要支持通过 Format 指定 raw、prometheus、otlp_metricv1、otlp_tracev1 等数据格式， 其它配置变更可查看【配置参数】表。如果没有特殊的需求，使用默认的v1即可。

enable: true
version: v2
inputs:
  - Type: service_kafka
    Version: 2.1.1
    Brokers: 
        - 172.xx.xx.48
        - 172.xx.xx.34
    ConsumerGroup: test-group
    Topics:
        - topicA
        - topicB
    ClientID: sls
flushers:
  - Type: flusher_stdout
    OnlyStdout: true  

• 输出

{"eventType":"byteArray","name":"","timestamp":0,"observedTimestamp":0,"tags":{},"byteArray":"{\"payload \": \"foo \"}"}