kafka

README

kafka输入源

该插件可从kafka采集数据到日志服务。

参数说明

插件类型（type）为 service_kafka。

参数	类型	必选或可选	参数说明
ConsumerGroup	string	必选	消费组名。
ClientID	string	必选	客户机ID。
Topics	string	必选	kafka主题。
Brokers	string	必选	kafka代理。
MaxMessageLen	int	可选	正整数，最大消息长度（1～524288）。
Version	string	可选	kafka版本，默认为空。
Offset	string	可选	初始偏移量（oldest，newest），默认oldest。
SASLUsername	string	可选	SASL用户名。
SASLPassword	string	可选	SASL密码。

示例

从kafka采集数据到日志服务，配置详情如下：

• 配置详情

{
  "inputs":[
    {
      "type":"service_kafka",
      "detail":{
        "Brokers": ["localhost:9092"],
        "Topics": ["test1"],
        "ConsumerGroup": "group1",
        "ClientID": "id1",
        "Offset": "oldest"
      }
    }
  ]
}

Documentation

Index

• type InputKafka
  • func (k *InputKafka) Collect(collector ilogtail.Collector) error
  • func (k *InputKafka) Description() string
  • func (k *InputKafka) Init(context ilogtail.Context) (int, error)
  • func (k *InputKafka) Start(collector ilogtail.Collector) error
  • func (k *InputKafka) Stop() error

Types

type InputKafka

type InputKafka struct {
	ConsumerGroup string
	ClientID      string
	Topics        []string
	Brokers       []string
	MaxMessageLen int
	Version       string
	Offset        string
	SASLUsername  string
	SASLPassword  string
	// contains filtered or unexported fields
}

func (*InputKafka) Collect

func (k *InputKafka) Collect(collector ilogtail.Collector) error

func (*InputKafka) Description

func (k *InputKafka) Description() string

func (*InputKafka) Init

func (k *InputKafka) Init(context ilogtail.Context) (int, error)

func (*InputKafka) Start

func (k *InputKafka) Start(collector ilogtail.Collector) error

func (*InputKafka) Stop

func (k *InputKafka) Stop() error