kafka

Kafka source用于接收Kafka数据。

Example

  1. sources:
  2. - type: kafka
  3. brokers: ["kafka1.kafka.svc:9092"]
  4. topic: log-*

brokers

字段类型是否必填默认值含义
brokersstring数组必填Kafka broker地址

topic

字段类型是否必填默认值含义
topicstring必填接收的topics,可使用正则来匹配多个topic

groupId

字段类型是否必填默认值含义
groupIdstring非必填loggieLoggie消费kafka的groupId

queueCapacity

字段类型是否必填默认值含义
queueCapacityint非必填100内部发送的队列容量

minAcceptedBytes

字段类型是否必填默认值含义
minAcceptedBytesint非必填1最小接收的batch字节数

maxAcceptedBytes

字段类型是否必填默认值含义
maxAcceptedBytesint非必填1e6(1MB)最大接收的消息字节数,如果超过会被truncate,可以设置为一个能容忍的较大的值

readMaxAttempts

字段类型是否必填默认值含义
readMaxAttemptsint非必填3最大的重试次数

maxPollWait

字段类型是否必填默认值含义
maxPollWaittime.Duration非必填10s接收的最长等待时间

readBackoffMin

字段类型是否必填默认值含义
readBackoffMintime.Duration非必填100ms在接收新的消息前,最小的时间间隔

readBackoffMax

字段类型是否必填默认值含义
readBackoffMaxtime.Duration非必填1s在接收新的消息前,最大的时间间隔

enableAutoCommit

字段类型是否必填默认值含义
enableAutoCommitbool非必填false是否开启autoCommit

autoCommitInterval

字段类型是否必填默认值含义
autoCommitIntervaltime.Duration非必填1sautoCommit的间隔时间

autoOffsetReset

字段类型是否必填默认值含义
autoOffsetResetstring非必填latest没有offset时,初始的offset采用方式,可为earliestlatest

sasl

字段类型是否必填默认值含义
sasl非必填SASL authentication
sasl.typestring必填SASL类型,可为:plainscram
sasl.userNamestring必填用户名
sasl.passwordstring必填密码
sasl.algorithmstringtype=scram时必填type=scram时使用的算法,可选sha256sha512