Bugfix: KafkaConsumer: reconnect on timeout

ynHuber · ynHuber · commit c99445564d7d · 2025-04-22T07:49:36.000+02:00
diff --git a/segments/input/kafkaconsumer/kafkaconsumer.go b/segments/input/kafkaconsumer/kafkaconsumer.go
@@ -261,7 +261,7 @@ func (segment *KafkaConsumer) Run(wg *sync.WaitGroup) {
 	defer func() {
 		handlerWg.Wait()
 		if err = client.Close(); err != nil {
-			log.Panic().Err(err).Msg("KafkaConsumer: Error closing Kafka client:")
+			log.Error().Err(err).Msg("KafkaConsumer: Error closing Kafka client:")
 		}
 	}()
 	go func() {
@@ -299,6 +299,13 @@ func (segment *KafkaConsumer) Run(wg *sync.WaitGroup) {
 			handlerCancel()
 			log.Info().Msg("KafkaConsumer: Connection Closed")
 			return
+		case handlerReady = <-handler.ready:
+			if !handlerReady {
+				log.Error().Msg("KafkaConsumer: Failed to establish connection.")
+				handlerCancel()
+				return
+			}
+			log.Info().Msg("KafkaConsumer: Reconnected and operational.")
 		}
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -261,7 +261,7 @@ func (segment KafkaConsumer) Run(wg sync.WaitGroup) {`
`261`	`261`	`defer func() {`
`262`	`262`	`handlerWg.Wait()`
`263`	`263`	`if err = client.Close(); err != nil {`
`264`		`- log.Panic().Err(err).Msg("KafkaConsumer: Error closing Kafka client:")`
	`264`	`+ log.Error().Err(err).Msg("KafkaConsumer: Error closing Kafka client:")`
`265`	`265`	`}`
`266`	`266`	`}()`
`267`	`267`	`go func() {`
`@@ -299,6 +299,13 @@ func (segment KafkaConsumer) Run(wg sync.WaitGroup) {`
`299`	`299`	`handlerCancel()`
`300`	`300`	`log.Info().Msg("KafkaConsumer: Connection Closed")`
`301`	`301`	`return`
	`302`	`+ case handlerReady = <-handler.ready:`
	`303`	`+ if !handlerReady {`
	`304`	`+ log.Error().Msg("KafkaConsumer: Failed to establish connection.")`
	`305`	`+ handlerCancel()`
	`306`	`+ return`
	`307`	`+ }`
	`308`	`+ log.Info().Msg("KafkaConsumer: Reconnected and operational.")`
`302`	`309`	`}`
`303`	`310`	`}`
`304`	`311`	`}`