DE-3386 | Redirecting litmus events to new topic

producer/producer_client.go

@@ -14,16 +14,18 @@ import (
 )
 
 var (
-	asyncProducer sarama.AsyncProducer
-	logger        *zap.Logger
-	cb            *gobreaker.CircuitBreaker
-	kafkaTopic    string
+	asyncProducer           sarama.AsyncProducer
+	logger                  *zap.Logger
+	cb                      *gobreaker.CircuitBreaker
+	defaultClickstreamTopic string
+	litmusClickstreamTopic  string
 )
 
 func InitializeProducers(kafkaConfiguration config.KafkaConfigurations, env string) {
 	logger = utils.GetLogger()
 	asyncProducer = GetAsyncProducer(kafkaConfiguration, env)
-	kafkaTopic = kafkaConfiguration.Kafka_Topic_Json
+	defaultClickstreamTopic = kafkaConfiguration.Kafka_Topic_Json
+	litmusClickstreamTopic = kafkaConfiguration.Kafka_Topic_Litmus
 
 	cb = gobreaker.NewCircuitBreaker(gobreaker.Settings{
 		Name:    "kafka-producer",
@@ -43,8 +45,12 @@ func InitializeProducers(kafkaConfiguration config.KafkaConfigurations, env stri
 	go processProducerErrors()
 }
 
-func GetKafkaTopic() string {
-	return kafkaTopic
+func GetDefaultClickstreamTopic() string {
+	return defaultClickstreamTopic
+}
+
+func GetLitmusClickstreamTopic() string {
+	return litmusClickstreamTopic
 }
 
 func GetAsyncProducerConfig(kafkaConfiguration config.KafkaConfigurations, env string) *sarama.Config {