janus/producer/kafka_writer.go

package producer

import (
	metrics "com.navi.medici.janus/instrumentation"
	"com.navi.medici.janus/utils"
	"github.com/Shopify/sarama"
	"go.uber.org/zap"
	"time"
)

func WriteMessageToKafkaAsync(message *sarama.ProducerMessage, source string) {
	beforeKafkaPushTime := utils.NanosToMillis(time.Now().UnixNano())
	asyncProducer.Input() <- message
	metrics.KafkaPushTimeHist.WithLabelValues(message.Topic, source).Observe(float64(utils.NanosToMillis(time.Now().UnixNano()) - beforeKafkaPushTime))
	metrics.KafkaPushCounter.WithLabelValues(message.Topic, source).Inc()
	logger.Info("SUCCESSFULLY WRITTEN TO KAFKA FOR SOURCE " + source)
}

func processProducerSuccesses() {
	for {
		msg := <-asyncProducer.Successes()
		metrics.ProducerSuccessCounter.WithLabelValues(msg.Topic).Inc()
	}
}

func processProducerErrors() {
	for {
		err := <-asyncProducer.Errors()
		logger.Error("Failed to write message to Kafka", zap.Error(err.Err))
		metrics.ProducerFailCounter.WithLabelValues(err.Msg.Topic).Inc()
	}

}