> For the complete documentation index, see [llms.txt](https://ilogtail.gitbook.io/ilogtail-docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://ilogtail.gitbook.io/ilogtail-docs/pre-release/plugins/shu-ru-cha-jian/kuo-zhan-shu-ru-cha-jian/service-kafka.md).
# Kafka
## 简介
`service_kafka` `input`插件实现了`ServiceInputV1`和`ServiceInputV2`接口，插件用于采集Kafka的消息。
## 事件能力
列含义见 [概览 · 事件能力列说明](https://ilogtail.gitbook.io/ilogtail-docs/pre-release/plugins/shu-ru-cha-jian/kuo-zhan-shu-ru-cha-jian/pages/V3l051tHPlzCBkmzOfvu#事件能力列说明)。
| Log | Metric | Span |
| --- | ------ | ---- |
| ✓ | — | — |
## 版本
[Stable](/ilogtail-docs/pre-release/plugins/stability-level.md)
## 版本说明
* 推荐版本：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，其余取默认值。
* 输入
```json
{"payload":"foo"}
```
### 采集配置（v1）
```yaml
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
```
* 输出
```json
{"payload":"foo"}
```
### 采集配置（v2）
`v2` 是自 **1.6.0** 起新增的实现（旧文档常写作 ilogtail），主要支持通过 `Format` 指定 `raw`、`prometheus`、`otlp_metricv1`、`otlp_tracev1` 等数据格式， 其它配置变更可查看【配置参数】表。如果没有特殊的需求，使用默认的`v1`即可。
```yaml
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
```
* 输出
```json
{"eventType":"byteArray","name":"","timestamp":0,"observedTimestamp":0,"tags":{},"byteArray":"{\"payload \": \"foo \"}"}
```