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

Uh oh!

logtail 插件 input service_kafka报错 #459

Uh oh!

{{title}}

Uh oh!

Replies: 2 comments

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Select a reply

Uh oh!

logtail 插件 input service_kafka报错 #459

Uh oh!

codingdancer Oct 13, 2022

Replies: 2 comments

Uh oh!

Uh oh!

codingdancer Oct 13, 2022 Author

Uh oh!

Uh oh!

mcy567 Oct 27, 2022

codingdancer
Oct 13, 2022

codingdancer
Oct 13, 2022
Author

mcy567
Oct 27, 2022