Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: upgrade go kafka client sarama version v1.42.2 to v1.43.3 #1772

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat: upgrade go kafka client sarama version v1.42.2 to v1.43.3 #1772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d55d956
feat: upgrade go kafka client sarama version v1.42.2 to v1.43.3
Sep 20, 2024
cca1b42
feat: update kafka producer MaxMessageBytes default size
Sep 23, 2024
c5e60b3
Merge branch 'alibaba:main' into upgrade-kafka-client
WillMeng Sep 23, 2024
f7e4370
feat: change kafka default protocol version 1.0.0 to 2.1.0
Sep 24, 2024
08295e1
feat: update kafka doc, update supported kafka version 3.8.0
Sep 24, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 52 additions & 52 deletions docs/cn/plugins/flusher/flusher-kafka_v2.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,58 +10,58 @@

## 配置参数

| 参数 | 类型 | 是否必选 | 说明 |
|---------------------------------------|----------|------|------------------------------------------------------------------------------------------------------------|
| Type | String | 是 | 插件类型 |
| Brokers | String数组 | 是 | Kafka Brokers |
| Topic | String | 是 | Kafka Topic,支持动态topic, 例如: `test_%{content.appname}` |
| Version | String | 否 | Kafka协议版本号 ,例如：`2.0.0`，默认值：`1.0.0` |
| Headers | header数组 | 否 | kafka消息头 ，配置使用请参考本文中`Headers`配置用例 |
| Convert | Struct | 否 | iLogtail数据转换协议配置 |
| Convert.Protocol | String | 否 | iLogtail数据转换协议，kafka flusher 可选值：`custom_single`,`custom_single_flatten`,`otlp_log_v1`。默认值：`custom_single` |
| Convert.Encoding | String | 否 | iLogtail flusher数据转换编码，可选值：`json`、`none`、`protobuf`，默认值：`json` |
| Convert.TagFieldsRename | Map | 否 | 对日志中tags中的json字段重命名 |
| Convert.ProtocolFieldsRename | Map | 否 | iLogtail日志协议字段重命名，可当前可重命名的字段：`contents`,`tags`和`time` |
| Authentication | Struct | 否 | Kafka连接访问认证配置，支持`SASL/PLAIN`，根据kafka服务端认证方式选择配置 |
| Authentication.PlainText.Username | String | 否 | PlainText认证用户名 |
| Authentication.PlainText.Password | String | 否 | PlainText认证密码 |
| Authentication.SASL.Username | String | 否 | SASL认证用户名 |
| Authentication.SASL.Password | String | 否 | SASL认证密码 |
| Authentication.Sasl.SaslMechanism | String | 否 | SASL认证，配置可选项：`PLAIN`、`SCRAM-SHA-256`、`SCRAM-SHA-512` |
| Authentication.TLS.Enabled | Boolean | 否 | 是否启用TLS安全连接, |
| Authentication.TLS.CAFile | String | 否 | TLS CA根证书文件路径 |
| Authentication.TLS.CertFile | String | 否 | TLS连接`kafka`证书文件路径 |
| Authentication.TLS.KeyFile | String | 否 | TLS连接`kafka`私钥文件路径 |
| Authentication.TLS.MinVersion | String | 否 | TLS支持协议最小版本，可选配置：`1.0, 1.1, 1.2, 1.3`,默认：`1.2` |
| Authentication.TLS.MaxVersion | String | 否 | TLS支持协议最大版本,可选配置：`1.0, 1.1, 1.2, 1.3`,默认采用：`crypto/tls`支持的版本，当前`1.3` |
| Authentication.TLS.InsecureSkipVerify | Boolean | 否 | 是否跳过TLS证书校验 |
| Authentication.Kerberos.ServiceName | String | 否 | 服务名称，例如：kafka |
| Authentication.Kerberos.UseKeyTab | Boolean | 否 | 是否采用keytab，配置此项后需要配置KeyTabPath，默认为：`false` |
| Authentication.Kerberos.Username | Boolean | 否 | UseKeyTab设置为`false`的情况下，需要指定用户名 |
| Authentication.Kerberos.Password | String | 否 | UseKeyTab设置为`false`的情况下，需要指定密码 |
| Authentication.Kerberos.Realm | String | 否 | kerberos认证管理域,大小写敏感 |
| Authentication.Kerberos.ConfigPath | Boolean | 否 | Kerberos krb5.conf |
| Authentication.Kerberos.KeyTabPath | String | 否 | keytab的路径 |
| PartitionerType | String | 否 | Partitioner类型。取值：`roundrobin`、`hash`、`random`。默认为：`random`。 |
| RequiredAcks | int | 否 | ACK的可靠等级.0=无响应,1=等待本地消息,-1=等待所有副本提交.默认1， |
| Compression | String | 否 | 压缩算法，可选值：`none`, `snappy`，`lz4`和`gzip`，默认值`none` |
| CompressionLevel | Int | 否 | 压缩级别，可选值：`1~9`，默认值：`4`,设置为`0`则禁用`Compression` |
| MaxMessageBytes | Int | 否 | 一个批次提交的大小限制，配置和`message.max.bytes`对应，默认值：`1000000` |
| MaxOpenRequests | Int | 否 | 一个连接允许的最大打开的请求数，默认值:`5` |
| MaxRetries | Int | 否 | 提交失败重试次数，最大`3`次，默认值：`3` |
| BulkMaxSize | Int | 否 | 单次请求提交事件数，默认`2048` |
| BulkFlushFrequency | Int | 否 | 发送批量 Kafka 请求之前等待的时间,0标识没有时延，默认值:`0` |
| Timeout | Int | 否 | 等待Kafka brokers响应的超时时间，默认`30s` |
| BrokerTimeout | int | 否 | kafka broker等待请求的最大时长，默认`10s` |
| Metadata.Retry.Max | int | 否 | 最大重试次数，默认值：`3` |
| Metadata.Retry.Backoff | int | 否 | 在重试之前等待leader选举发生的时间，默认值：`250ms` |
| Metadata.RefreshFrequency | int | 否 | Metadata刷新频率，默认值：`250ms` |
| Metadata.Full | int | 否 | 获取原数数据的策略，获取元数据时使用的策略，当此选项为`true`时，客户端将为所有可用主题维护一整套元数据，如果此选项设置为`false`，它将仅刷新已配置主题的元数据。默认值:`false`。 |
| HashKeys | String数组 | 否 | PartitionerType为`hash`时，需指定HashKeys。 |
| HashOnce | Boolean | 否 | |
| ClientID | String | 否 | 写入Kafka的Client ID，默认取值：`LogtailPlugin`。 |

- `Version`需要填写的是`kafka protocol version`版本号，`flusher_kafka_v2`当前支持的`kafka`版本范围：`0.8.2.x~3.3.1`。
| 参数 | 类型 | 是否必选 | 说明 |
| ------------------------------------- | ---------- | -------- | ------------------------------------------------------------ |
| Type | String | 是 | 插件类型 |
| Brokers | String数组 | 是 | Kafka Brokers |
| Topic | String | 是 | Kafka Topic,支持动态topic, 例如: `test_%{content.appname}` |
| Version | String | 否 | Kafka协议版本号 ,例如：`2.1.0`，默认值：`2.1.0` |
| Headers | header数组 | 否 | kafka消息头 ，配置使用请参考本文中`Headers`配置用例 |
| Convert | Struct | 否 | iLogtail数据转换协议配置 |
| Convert.Protocol | String | 否 | iLogtail数据转换协议，kafka flusher 可选值：`custom_single`,`custom_single_flatten`,`otlp_log_v1`。默认值：`custom_single` |
| Convert.Encoding | String | 否 | iLogtail flusher数据转换编码，可选值：`json`、`none`、`protobuf`，默认值：`json` |
| Convert.TagFieldsRename | Map | 否 | 对日志中tags中的json字段重命名 |
| Convert.ProtocolFieldsRename | Map | 否 | iLogtail日志协议字段重命名，可当前可重命名的字段：`contents`,`tags`和`time` |
| Authentication | Struct | 否 | Kafka连接访问认证配置，支持`SASL/PLAIN`，根据kafka服务端认证方式选择配置 |
| Authentication.PlainText.Username | String | 否 | PlainText认证用户名 |
| Authentication.PlainText.Password | String | 否 | PlainText认证密码 |
| Authentication.SASL.Username | String | 否 | SASL认证用户名 |
| Authentication.SASL.Password | String | 否 | SASL认证密码 |
| Authentication.Sasl.SaslMechanism | String | 否 | SASL认证，配置可选项：`PLAIN`、`SCRAM-SHA-256`、`SCRAM-SHA-512` |
| Authentication.TLS.Enabled | Boolean | 否 | 是否启用TLS安全连接, |
| Authentication.TLS.CAFile | String | 否 | TLS CA根证书文件路径 |
| Authentication.TLS.CertFile | String | 否 | TLS连接`kafka`证书文件路径 |
| Authentication.TLS.KeyFile | String | 否 | TLS连接`kafka`私钥文件路径 |
| Authentication.TLS.MinVersion | String | 否 | TLS支持协议最小版本，可选配置：`1.0, 1.1, 1.2, 1.3`,默认：`1.2` |
| Authentication.TLS.MaxVersion | String | 否 | TLS支持协议最大版本,可选配置：`1.0, 1.1, 1.2, 1.3`,默认采用：`crypto/tls`支持的版本，当前`1.3` |
| Authentication.TLS.InsecureSkipVerify | Boolean | 否 | 是否跳过TLS证书校验 |
| Authentication.Kerberos.ServiceName | String | 否 | 服务名称，例如：kafka |
| Authentication.Kerberos.UseKeyTab | Boolean | 否 | 是否采用keytab，配置此项后需要配置KeyTabPath，默认为：`false` |
| Authentication.Kerberos.Username | Boolean | 否 | UseKeyTab设置为`false`的情况下，需要指定用户名 |
| Authentication.Kerberos.Password | String | 否 | UseKeyTab设置为`false`的情况下，需要指定密码 |
| Authentication.Kerberos.Realm | String | 否 | kerberos认证管理域,大小写敏感 |
| Authentication.Kerberos.ConfigPath | Boolean | 否 | Kerberos krb5.conf |
| Authentication.Kerberos.KeyTabPath | String | 否 | keytab的路径 |
| PartitionerType | String | 否 | Partitioner类型。取值：`roundrobin`、`hash`、`random`。默认为：`random`。 |
| RequiredAcks | int | 否 | ACK的可靠等级.0=无响应,1=等待本地消息,-1=等待所有副本提交.默认1， |
| Compression | String | 否 | 压缩算法，可选值：`none`, `snappy`，`lz4`和`gzip`，默认值`none` |
| CompressionLevel | Int | 否 | 压缩级别，可选值：`1~9`，默认值：`4`,设置为`0`则禁用`Compression` |
| MaxMessageBytes | Int | 否 | 一个批次提交的大小限制，配置和`message.max.bytes`对应，默认值：`1048576` |
| MaxOpenRequests | Int | 否 | 一个连接允许的最大打开的请求数，默认值:`5` |
| MaxRetries | Int | 否 | 提交失败重试次数，最大`3`次，默认值：`3` |
| BulkMaxSize | Int | 否 | 单次请求提交事件数，默认`2048` |
| BulkFlushFrequency | Int | 否 | 发送批量 Kafka 请求之前等待的时间,0标识没有时延，默认值:`0` |
| Timeout | Int | 否 | 等待Kafka brokers响应的超时时间，默认`30s` |
| BrokerTimeout | int | 否 | kafka broker等待请求的最大时长，默认`10s` |
| Metadata.Retry.Max | int | 否 | 最大重试次数，默认值：`3` |
| Metadata.Retry.Backoff | int | 否 | 在重试之前等待leader选举发生的时间，默认值：`250ms` |
| Metadata.RefreshFrequency | int | 否 | Metadata刷新频率，默认值：`250ms` |
| Metadata.Full | int | 否 | 获取原数数据的策略，获取元数据时使用的策略，当此选项为`true`时，客户端将为所有可用主题维护一整套元数据，如果此选项设置为`false`，它将仅刷新已配置主题的元数据。默认值:`false`。 |
| HashKeys | String数组 | 否 | PartitionerType为`hash`时，需指定HashKeys。 |
| HashOnce | Boolean | 否 | |
| ClientID | String | 否 | 写入Kafka的Client ID，默认取值：`LogtailPlugin`。 |

- `Version`需要填写的是`kafka protocol version`版本号，`flusher_kafka_v2`当前支持的`kafka`版本范围：`0.8.2.x~3.8.0`。
请根据自己的`kafka`版本号参照下面的`kafka protocol version`规则进行配置。**建议根据自己的`kafka`
版本指定对应`protocol version`**,
`kafka protocol version`支持版本号如下：
Expand Down
Loading
Loading