fix(kafkaclient): do not close logs channel (#180)

Author: smoya

kafka/config.go

@@ -1,7 +1,6 @@
 package kafka
 
 import (
-	"context"
 	"fmt"
 	"log/syslog"
 	"strings"
@@ -153,7 +152,7 @@ func (c Config) configureAuth(configMap *kafkalib.ConfigMap) error {
 type ConfigOpt func(c *kafkalib.ConfigMap)
 
 // WithLogger adds a logger to a Kafka consumer or producer.
-func WithLogger(ctx context.Context, log logrus.FieldLogger) ConfigOpt {
+func WithLogger(log logrus.FieldLogger) ConfigOpt {
 	return func(c *kafkalib.ConfigMap) {
 
 		syslogToLogrusLevelMapping := map[syslog.Priority]logrus.Level{
@@ -175,27 +174,20 @@ func WithLogger(ctx context.Context, log logrus.FieldLogger) ConfigOpt {
 
 		// Read from channel and print logs using the provided logger.
 		go func() {
-			defer close(logsChan)
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case m, ok := <-logsChan:
-					if !ok {
-						return
-					}
-					l := log.WithFields(logrus.Fields{
-						"kafka_context": m.Tag,
-						"kafka_client":  m.Name,
-					}).WithTime(m.Timestamp)
-
-					logrusLevel := syslogToLogrusLevelMapping[syslog.Priority(m.Level)]
-					switch logrusLevel {
-					case logrus.ErrorLevel:
-						l.WithError(errors.New(m.Message)).Error("Error in Kafka Consumer")
-					default:
-						l.Log(logrusLevel, m.Message)
-					}
+			// Do not close logsChan because confluent-kafka-go will send logs until we close the client.
+			// Otherwise it will panic trying to send messages to a closed channel.
+			for m := range logsChan {
+				l := log.WithFields(logrus.Fields{
+					"kafka_context": m.Tag,
+					"kafka_client":  m.Name,
+				}).WithTime(m.Timestamp)
+
+				logrusLevel := syslogToLogrusLevelMapping[syslog.Priority(m.Level)]
+				switch logrusLevel {
+				case logrus.ErrorLevel:
+					l.WithError(errors.New(m.Message)).Error("Error in Kafka Consumer")
+				default:
+					l.Log(logrusLevel, m.Message)
 				}
 			}
 		}()

kafka/integration_test.go

@@ -50,7 +50,7 @@ func TestIntegration(t *testing.T) {
 		}
 
 		// create the producer
-		p, err := NewProducer(conf, WithLogger(ctx, log), WithPartitionerAlgorithm(PartitionerMurMur2))
+		p, err := NewProducer(conf, WithLogger(log), WithPartitionerAlgorithm(PartitionerMurMur2))
 		assert.NoError(err)
 		assert.NotNil(p)
 
@@ -131,7 +131,7 @@ func TestIntegration(t *testing.T) {
 		val := "gotestval"
 
 		// create the producer
-		p, err := NewProducer(conf, WithLogger(ctx, log), WithPartitionerAlgorithm(PartitionerMurMur2))
+		p, err := NewProducer(conf, WithLogger(log), WithPartitionerAlgorithm(PartitionerMurMur2))
 		assert.NoError(err)
 		assert.NotNil(p)