kafka¶

Kafka source is used for receice Kafka data.

Example

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

brokers¶

`field`	`type`	`required`	`default`	`description`
brokers	string array	true	none	Kafka broker address

`field`	`type`	`required`	`default`	`description`
topic	string	true	none	receiving topics. You can use regular expressions to match multiple topics

`field`	`type`	`required`	`default`	`description`
groupId	string	false	loggie	groupId Loggie use to consume kafka

`field`	`type`	`required`	`default`	`description`
queueCapacity	int	false	100	capacity of internal sending queue

`field`	`type`	`required`	`default`	`description`
minAcceptedBytes	int	false	1	minimum bytes of received batch

`field`	`type`	`required`	`default`	`description`
maxAcceptedBytes	int	false	1e6	maximum bytes of received batch

`field`	`type`	`required`	`default`	`description`
readMaxAttempts	int	false	3	maximum number of retries

`field`	`type`	`required`	`default`	`description`
maxPollWait	time.Duration	false	10s	maximum time waiting to receive

`field`	`type`	`required`	`default`	`description`
readBackoffMin	time.Duration	false	100ms	minimum time interval before receiving a new message

`field`	`type`	`required`	`default`	`description`
readBackoffMax	time.Duration	false	1s	maximum time interval before receiving a new message

`field`	`type`	`required`	`default`	`description`
enableAutoCommit	bool	false	false	whether to enable autoCommit

`field`	`type`	`required`	`default`	`description`
autoCommitInterval	time.Duration	false	1s	Interval time for autoCommit

`field`	`type`	`required`	`default`	`description`
autoOffsetReset	string	false	latest	the initial offset adopted when there is no offset, which can be `earliest` or `latest`

`field`	`type`	`required`	`description`
sasl		false	SASL authentication
sasl.type	string	false	SASL type, which can be `plain`, `scram`
sasl.userName	string	false	username
sasl.password	string	false	password
sasl.algorithm	string	true when type=scram	Algorithm to use when type=scram. `sha256` or `sha512`