From a741af7c9cb85c5304c461acfc76746ec798ff19 Mon Sep 17 00:00:00 2001
From: varnit goyal <github.cicd@navi.com>
Date: Fri, 11 Oct 2024 00:06:51 +0530
Subject: [PATCH] TP-5555 | kafka fix

---
 configs/application.yml | 1 +
 configs/kafka.go        | 2 ++
 pkg/kafka/config.go     | 2 +-
 3 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/configs/application.yml b/configs/application.yml
index 0ebcdd1..38f1c39 100644
--- a/configs/application.yml
+++ b/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
diff --git a/configs/kafka.go b/configs/kafka.go
index 09a637e..5b488f2 100644
--- a/configs/kafka.go
+++ b/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"),
 	}
 }
 
diff --git a/pkg/kafka/config.go b/pkg/kafka/config.go
index 2eb5cab..09b6961 100644
--- a/pkg/kafka/config.go
+++ b/pkg/kafka/config.go
@@ -9,7 +9,7 @@ func ConfluentProducer(baseConfig *configs.KafkaConfig) (*kafka.Producer, error)
 	return kafka.NewProducer(&kafka.ConfigMap{
 		"bootstrap.servers": baseConfig.Brokers[0],
 		"security.protocol": "SASL_SSL",
-		"sasl.mechanisms":   "PLAIN",
+		"sasl.mechanisms":   baseConfig.Mechanism,
 		"sasl.username":     baseConfig.Username,
 		"sasl.password":     baseConfig.Password,
 	})