Fix: Kafka consumer fails to reconnect and sensor stops working (#3495)

smunukutla-mycarrier · web-flow · commit 3595e84e5a79 · 2025-03-08T10:14:59.000-08:00
Signed-off-by: Siva Munukutla &lt;smunukutla@mycarrier.io&gt;
diff --git a/pkg/eventbus/kafka/sensor/kafka_sensor.go b/pkg/eventbus/kafka/sensor/kafka_sensor.go
@@ -219,6 +219,11 @@ func (s *KafkaSensor) Listen(ctx context.Context) {
 				)
 			}
 
+			// Fail fast if the server returns an incomplete response
+			if err == sarama.ErrIncompleteResponse {
+				s.Logger.Fatalw("Kafka error", zap.Error(err))
+			}
+
 			s.Logger.Errorw("Failed to consume", zap.Error(err))
 			return
 		}

Original file line number	Diff line number	Diff line change
`@@ -219,6 +219,11 @@ func (s *KafkaSensor) Listen(ctx context.Context) {`
`219`	`219`	`)`
`220`	`220`	`}`
`221`	`221`
	`222`	`+ // Fail fast if the server returns an incomplete response`
	`223`	`+ if err == sarama.ErrIncompleteResponse {`
	`224`	`+ s.Logger.Fatalw("Kafka error", zap.Error(err))`
	`225`	`+ }`
	`226`	`+`
`222`	`227`	`s.Logger.Errorw("Failed to consume", zap.Error(err))`
`223`	`228`	`return`
`224`	`229`	`}`