add session-timeout, heartbeat-interval and recovery-wait-time to kafka_input
@@ -23,6 +23,8 @@ const (
defaultTopic = "telemetry"
defaultNumWorkers = 1
defaultSessionTimeout = 10 * time.Second
+ defaultHeartbeatInterval = 3 * time.Second
+ defaultRecoveryWaitTime = 2 * time.Second
defaultAddress = "localhost:9092"
defaultGroupID = "gnmic-consumers"
)
@@ -54,8 +56,8 @@ type Config struct {
Address string `mapstructure:"address,omitempty"`
Topic string `mapstructure:"topic,omitempty"`
GroupID string `mapstructure:"group-id,omitempty"`
- MaxRetry int `mapstructure:"max-retry,omitempty"`
- Timeout time.Duration `mapstructure:"timeout,omitempty"`
+ SessionTimeout time.Duration `mapstructure:"session-timeout,omitempty"`
+ HeartbeatInterval time.Duration `mapstructure:"heartbeat-interval,omitempty"`
RecoveryWaitTime time.Duration `mapstructure:"recovery-wait-time,omitempty"`
Version string `mapstructure:"version,omitempty"`
Format string `mapstructure:"format,omitempty"`
@@ -88,10 +90,11 @@ func (k *KafkaInput) Start(ctx context.Context, cfg map[string]interface{}, opts
func (k *KafkaInput) worker(ctx context.Context, idx int) {
defer k.wg.Done()
config := sarama.NewConfig()
- config.Version = sarama.V0_11_0_1
+ config.Version = k.Cfg.kafkaVersion
config.ClientID = fmt.Sprintf("%s-%d", k.Cfg.Name, idx)
config.Consumer.Return.Errors = true
- config.Consumer.Group.Session.Timeout = k.Cfg.Timeout
+ config.Consumer.Group.Session.Timeout = k.Cfg.SessionTimeout
+ config.Consumer.Group.Heartbeat.Interval = k.Cfg.HeartbeatInterval
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
// TODO: finish kafka config
@@ -232,12 +235,18 @@ func (k *KafkaInput) setDefaults() error {
if k.Cfg.NumWorkers <= 0 {
k.Cfg.NumWorkers = defaultNumWorkers
}
- if k.Cfg.Timeout <= 2*time.Millisecond {
- k.Cfg.Timeout = defaultSessionTimeout
+ if k.Cfg.SessionTimeout <= 2*time.Millisecond {
+ k.Cfg.SessionTimeout = defaultSessionTimeout
+ }
+ if k.Cfg.HeartbeatInterval <= 1*time.Millisecond {
+ k.Cfg.HeartbeatInterval = defaultHeartbeatInterval
}
if k.Cfg.GroupID == "" {
k.Cfg.GroupID = defaultGroupID
}
+ if k.Cfg.RecoveryWaitTime <= 0 {
+ k.Cfg.RecoveryWaitTime = defaultRecoveryWaitTime
+ }
return nil
}
