Merge pull request #6 from navi-ppl/TP-5555/clean-up

TP-87883 | cybertron log enricher pipeline

configs/application.yml

@@ -53,6 +53,7 @@ kafka:
     enabled: false
   sasl:
     enabled: true
+    mechanism: KAFKA_SASL_MECHANISM
   consumer:
     batch.size: 1
     goroutines.max.pool.size: 20

configs/kafka.go

@@ -17,6 +17,7 @@ type KafkaConfig struct {
 	ConsumerBatchSize         int
 	GoroutinesMaxPoolSize     int
 	ConsumerMaxTimeoutMs      string
+	Mechanism                 string
 }
 
 func NewKafkaConfig() *KafkaConfig {
@@ -32,6 +33,7 @@ func NewKafkaConfig() *KafkaConfig {
 		ConsumerBatchSize:         viper.GetInt("kafka.consumer.batch.size"),
 		GoroutinesMaxPoolSize:     viper.GetInt("kafka.consumer.goroutines.max.pool.size"),
 		ConsumerMaxTimeoutMs:      viper.GetString("kafka.consumer.max.timeout.ms"),
+		Mechanism:                 viper.GetString("kafka.sasl.mechanism"),
 	}
 }
 

pkg/kafka/consumer/consume.go

@@ -22,7 +22,7 @@ func NewKafkaConsumers(baseConfig *configs.KafkaConfig, errorProcessor *service.
 		"session.timeout.ms": 6000,
 		"auto.offset.reset":  "earliest",
 		"security.protocol":  "SASL_SSL",
-		"sasl.mechanisms":    "PLAIN",
+		"sasl.mechanisms":    baseConfig.Mechanism,
 		"sasl.username":      baseConfig.Username,
 		"sasl.password":      baseConfig.Password,
 	})