DE-121 - added scram

producer/producer_config.go

@@ -8,6 +8,7 @@ import (
 	"time"
 	"strings"
 	"github.com/Shopify/sarama"
+	"crypto/tls"
 	// "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
 )
 
@@ -22,6 +23,8 @@ func GetSyncProducerConfig(kafkaConfiguration config.KafkaConfigurations, env st
 	// security configs
 	// config.Net.TLS.Enable             = true
 	if env == "PROD" {
+	    config.Net.TLS.Enable             = true
+    	config.Net.TLS.Config             = createTLSConfiguration()
 	    config.Net.SASL.Enable            = true
         config.Net.SASL.Handshake         = true
         config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
@@ -53,6 +56,8 @@ func GetAsyncProducerConfig(kafkaConfiguration config.KafkaConfigurations, env s
 	// security configs
 	// config.Net.TLS.Enable             = true
 	if env == "PROD" {
+	    config.Net.TLS.Enable             = true
+	    config.Net.TLS.Config             = createTLSConfiguration()
         config.Net.SASL.Enable            = true
         config.Net.SASL.Handshake         = true
         config.Net.SASL.Mechanism         = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
@@ -105,6 +110,14 @@ func InitializeProducers(kafkaConfiguration config.KafkaConfigurations, env stri
 	asyncProducer = GetAsyncProducer(kafkaConfiguration, env)
 }
 
+func createTLSConfiguration() (t *tls.Config) {
+	t = &tls.Config{
+		InsecureSkipVerify: false,
+	}
+	return t
+}
+
+
 // using confluent-kafka-go
 // func GetProducerConfig(kafkaConfiguration config.KafkaConfigurations) *kafka.ConfigMap {
 // 	var config = kafka.ConfigMap {