logtail 插件 input service_kafka报错 #459

codingdancer · 2022-10-13T03:33:01Z

codingdancer
Oct 13, 2022

docker环境下采集kakfa 数据，发现如下问题：
1.topic 配置问题
应该使用key 为Topics 配置，否则会报找不到topic
2.sarama-cluster 消费问题
具体报错信息

D:\software\dockerdesktop\ilogtail\1.1.1>docker logs docker_ilogtail
ilogtail started. pid: 12
register fun v2 0x697dd0 0x699a90 0x699490 0x699ac0
2022/10/13 03:20:43 DEBUG Now using Go's stdlib log package (via loggers/mappers/stdlib).
load log config /usr/local/ilogtail/plugin_logger.xml
panic: non-positive interval for NewTicker

goroutine 210 [running]:
time.NewTicker(0x0, 0x7f5427a69438)
        /go/src/time/tick.go:24 +0x151
github.com/bsm/sarama-cluster.(*Consumer).cmLoop(0xc0002b42d0, 0xc00066f4a0)
        /src/vendor/github.com/bsm/sarama-cluster/consumer.go:452 +0x5c
github.com/bsm/sarama-cluster.(*loopTomb).Go.func1(0xc000145360, 0xc00021af70)
        /src/vendor/github.com/bsm/sarama-cluster/util.go:73 +0x7d
created by github.com/bsm/sarama-cluster.(*loopTomb).Go
        /src/vendor/github.com/bsm/sarama-cluster/util.go:69 +0x68
ilogtail exited unexpectedly

类似： https://blog.csdn.net/ecjtusanhu/article/details/106376699

docker 配置如下：

enable: true
inputs:
  - Type: service_kafka
    Version: 2.1.1
    Brokers: 
      - 10.71.48.62:9092
    ConsumerGroup: skyeyeResult
    Topics:
      - test
    ClientID: ilogtail
flushers:   
  - Type: flusher_kafka
    Brokers: 
      - 10.71.48.62:9092
    Topic: bmw_bms_collector_rsk

Answered by codingdancer

Oct 13, 2022

docker环境下采集kakfa 数据，发现如下问题： 1.topic 配置问题应该使用key 为Topics 配置，否则会报找不到topic 2.sarama-cluster 消费问题具体报错信息

D:\software\dockerdesktop\ilogtail\1.1.1>docker logs docker_ilogtail
ilogtail started. pid: 12
register fun v2 0x697dd0 0x699a90 0x699490 0x699ac0
2022/10/13 03:20:43 DEBUG Now using Go's stdlib log package (via loggers/mappers/stdlib).
load log config /usr/local/ilogtail/plugin_logger.xml
panic: non-positive interval for NewTicker

goroutine 210 [running]:
time.NewTicker(0x0, 0x7f5427a69438)
        /go/src/time/tick.go:24 +0x151
github.com/bsm/sarama-cluster.(*Consumer).cmLoop(0xc0002b42d0, 0xc00066f4a0)
        /src/vendor/github.com/bsm/sarama-cluster/consumer.go:452 +0x5c
github.c…

View full answer

codingdancer · 2022-10-13T06:30:08Z

codingdancer
Oct 13, 2022
Author

docker环境下采集kakfa 数据，发现如下问题： 1.topic 配置问题应该使用key 为Topics 配置，否则会报找不到topic 2.sarama-cluster 消费问题具体报错信息

D:\software\dockerdesktop\ilogtail\1.1.1>docker logs docker_ilogtail
ilogtail started. pid: 12
register fun v2 0x697dd0 0x699a90 0x699490 0x699ac0
2022/10/13 03:20:43 DEBUG Now using Go's stdlib log package (via loggers/mappers/stdlib).
load log config /usr/local/ilogtail/plugin_logger.xml
panic: non-positive interval for NewTicker

goroutine 210 [running]:
time.NewTicker(0x0, 0x7f5427a69438)
        /go/src/time/tick.go:24 +0x151
github.com/bsm/sarama-cluster.(*Consumer).cmLoop(0xc0002b42d0, 0xc00066f4a0)
        /src/vendor/github.com/bsm/sarama-cluster/consumer.go:452 +0x5c
github.com/bsm/sarama-cluster.(*loopTomb).Go.func1(0xc000145360, 0xc00021af70)
        /src/vendor/github.com/bsm/sarama-cluster/util.go:73 +0x7d
created by github.com/bsm/sarama-cluster.(*loopTomb).Go
        /src/vendor/github.com/bsm/sarama-cluster/util.go:69 +0x68
ilogtail exited unexpectedly

类似： https://blog.csdn.net/ecjtusanhu/article/details/106376699

docker 配置如下：

enable: true
inputs:
  - Type: service_kafka
    Version: 2.1.1
    Brokers: 
      - 10.71.48.62:9092
    ConsumerGroup: skyeyeResult
    Topics:
      - test
    ClientID: ilogtail
flushers:   
  - Type: flusher_kafka
    Brokers: 
      - 10.71.48.62:9092
    Topic: bmw_bms_collector_rsk

报错代码为，取出来的值为非正报错

ticker := time.NewTicker(c.client.config.Consumer.Offsets.CommitInterval)

看源码发现最新镜像使用的sarama 中已经丢弃CommitInterval字段，使用Consumer.Offsets.AutoCommit.Interval替代

	if c.Consumer.Offsets.CommitInterval != 0 {
		Logger.Println("Deprecation warning: Consumer.Offsets.CommitInterval exists for historical compatibility" +
			" and should not be used. Please use Consumer.Offsets.AutoCommit, the current value will be ignored")
	}

		// AutoCommit specifies configuration for commit messages automatically.
			AutoCommit struct {
				// Whether or not to auto-commit updated offsets back to the broker.
				// (default enabled).
				Enable bool

				// How frequently to commit updated offsets. Ineffective unless
				// auto-commit is enabled (default 1s)
				Interval time.Duration
			}

在初始配置时并未给默认值,只给c.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second

func NewConfig() *Config {
	c := &Config{}

	c.Admin.Retry.Max = 5
	c.Admin.Retry.Backoff = 100 * time.Millisecond
	c.Admin.Timeout = 3 * time.Second

	c.Net.MaxOpenRequests = 5
	c.Net.DialTimeout = 30 * time.Second
	c.Net.ReadTimeout = 30 * time.Second
	c.Net.WriteTimeout = 30 * time.Second
	c.Net.SASL.Handshake = true
	c.Net.SASL.Version = SASLHandshakeV0

	c.Metadata.Retry.Max = 3
	c.Metadata.Retry.Backoff = 250 * time.Millisecond
	c.Metadata.RefreshFrequency = 10 * time.Minute
	c.Metadata.Full = true

	c.Producer.MaxMessageBytes = 1000000
	c.Producer.RequiredAcks = WaitForLocal
	c.Producer.Timeout = 10 * time.Second
	c.Producer.Partitioner = NewHashPartitioner
	c.Producer.Retry.Max = 3
	c.Producer.Retry.Backoff = 100 * time.Millisecond
	c.Producer.Return.Errors = true
	c.Producer.CompressionLevel = CompressionLevelDefault

	c.Consumer.Fetch.Min = 1
	c.Consumer.Fetch.Default = 1024 * 1024
	c.Consumer.Retry.Backoff = 2 * time.Second
	c.Consumer.MaxWaitTime = 250 * time.Millisecond
	c.Consumer.MaxProcessingTime = 100 * time.Millisecond
	c.Consumer.Return.Errors = false
	c.Consumer.Offsets.AutoCommit.Enable = true
	c.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second
	c.Consumer.Offsets.Initial = OffsetNewest
	c.Consumer.Offsets.Retry.Max = 3

	c.Consumer.Group.Session.Timeout = 10 * time.Second
	c.Consumer.Group.Heartbeat.Interval = 3 * time.Second
	c.Consumer.Group.Rebalance.Strategy = BalanceStrategyRange
	c.Consumer.Group.Rebalance.Timeout = 60 * time.Second
	c.Consumer.Group.Rebalance.Retry.Max = 4
	c.Consumer.Group.Rebalance.Retry.Backoff = 2 * time.Second

	c.ClientID = defaultClientID
	c.ChannelBufferSize = 256
	c.Version = DefaultVersion
	c.MetricRegistry = metrics.NewRegistry()

	return c
}

解决方式： non-positive interva for newticker解决

0 replies

mcy567 · 2022-10-27T07:23:08Z

mcy567
Oct 27, 2022

遇到了同样的问题，不太会编译，下次版本出包得下个月底了吧

register fun v2 0x669cb0 0x66bab0 0x66b4b0 0x66bae0
2022/10/27 07:12:29 DEBUG Now using Go's stdlib log package (via loggers/mappers/stdlib).
load log config /opt/ilogtail/plugin_logger.xml 
recover stderr
recover stdout
panic: non-positive interval for NewTicker

goroutine 74 [running]:
time.NewTicker(0x7f5e3ca24149?)
	/go/src/time/tick.go:24 +0x10f
github.com/bsm/sarama-cluster.(*Consumer).cmLoop(0xc0001aa3c0, 0xc0004269c0)
	/src/vendor/github.com/bsm/sarama-cluster/consumer.go:452 +0x5c
github.com/bsm/sarama-cluster.(*loopTomb).Go.func1()
	/src/vendor/github.com/bsm/sarama-cluster/util.go:73 +0x88
created by github.com/bsm/sarama-cluster.(*loopTomb).Go
	/src/vendor/github.com/bsm/sarama-cluster/util.go:69 +0x8d

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

logtail 插件 input service_kafka报错 #459

{{title}}

Replies: 2 comments

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Select a reply

logtail 插件 input service_kafka报错 #459

codingdancer Oct 13, 2022

Replies: 2 comments

codingdancer Oct 13, 2022 Author

mcy567 Oct 27, 2022

codingdancer
Oct 13, 2022

codingdancer
Oct 13, 2022
Author

mcy567
Oct 27, 2022