refactor: remove the global variable and atomic update

Author: siddhanta-rath

.env.sample

@@ -26,8 +26,7 @@ PUBLISHER_KAFKA_CLIENT_RETRY_BACKOFF_MS="100"
 PUBLISHER_KAFKA_CLIENT_STATISTICS_INTERVAL_MS=5000
 PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES=100000
 PUBLISHER_KAFKA_FLUSH_INTERVAL_MS=1000
-PUBLISHER_KAFKA_DELIVERY_REPORT_INTERVAL_MS=60000
-PUBLISHER_KAFKA_DELIVERY_REPORT_TOPIC_NAME="clickstream-total-event-log"
+PUBLISHER_KAFKA_CLICKSTREAM_STATS_TOPIC_NAME="clickstream-total-event-log"
 
 METRIC_STATSD_ADDRESS=":8125"
 METRIC_STATSD_FLUSH_PERIOD_MS=100

.env.test

@@ -27,8 +27,7 @@ PUBLISHER_KAFKA_CLIENT_RETRY_BACKOFF_MS="100"
 PUBLISHER_KAFKA_CLIENT_STATISTICS_INTERVAL_MS=5000
 PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES=100000
 PUBLISHER_KAFKA_FLUSH_INTERVAL_MS=1000
-PUBLISHER_KAFKA_DELIVERY_REPORT_INTERVAL_MS=60000
-PUBLISHER_KAFKA_DELIVERY_REPORT_TOPIC_NAME="clickstream-total-event-log"
+PUBLISHER_KAFKA_CLICKSTREAM_STATS_TOPIC_NAME="clickstream-total-event-log"
 
 METRIC_STATSD_ADDRESS=":8125"
 METRIC_STATSD_FLUSH_PERIOD_MS=100

app/server.go

@@ -37,7 +37,6 @@ func StartServer(ctx context.Context, cancel context.CancelFunc, shutdown chan b
 	workerPool.StartWorkers()
 	go kPublisher.ReportStats()
 	go reportProcMetrics()
-	go kPublisher.ReportDeliveryEventCount()
 	// create signal channel at startup
 	signalChan := make(chan os.Signal, 1)
 	signal.Notify(signalChan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

config/load_test.go

@@ -63,14 +63,13 @@ func TestDynamicConfigLoad(t *testing.T) {
 
 func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	os.Setenv("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS", "1000")
-	os.Setenv("PUBLISHER_KAFKA_DELIVERY_REPORT_TOPIC_NAME", "clickstream-test-log")
+	os.Setenv("PUBLISHER_KAFKA_CLICKSTREAM_STATS_TOPIC_NAME", "clickstream-test-log")
 	os.Setenv("PUBLISHER_KAFKA_DELIVERY_REPORT_INTERVAL_MS", "60")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_BOOTSTRAP_SERVERS", "kafka:9092")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_ACKS", "1")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "10000")
 	os.Setenv("SOMETHING_PUBLISHER_KAFKA_CLIENT_SOMETHING", "anything")
 	publisherKafkaConfigLoader()
-	expectedDeliveryReportInterval := 60 * time.Millisecond
 	kafkaConfig := PublisherKafka.ToKafkaConfigMap()
 	bootstrapServer, _ := kafkaConfig.Get("bootstrap.servers", "")
 	topic, _ := kafkaConfig.Get("topic", "")
@@ -79,8 +78,7 @@ func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	assert.Equal(t, "", topic)
 	assert.NotEqual(t, something, "anything")
 	assert.Equal(t, 4, len(*kafkaConfig))
-	assert.Equal(t, expectedDeliveryReportInterval, PublisherKafka.DeliveryReportInterval)
-	assert.Equal(t, "clickstream-test-log", PublisherKafka.DeliveryReportTopic)
+	assert.Equal(t, "clickstream-test-log", PublisherKafka.ClickstreamStatsTopicName)
 }
 
 func TestWorkerConfig(t *testing.T) {

config/publisher.go

@@ -15,9 +15,9 @@ var PublisherKafka publisherKafka
 var dynamicKafkaClientConfigPrefix = "PUBLISHER_KAFKA_CLIENT_"
 
 type publisherKafka struct {
-	FlushInterval          int
-	DeliveryReportInterval time.Duration
-	DeliveryReportTopic    string
+	FlushInterval             int
+	DeliveryReportInterval    time.Duration
+	ClickstreamStatsTopicName string
 }
 
 func (k publisherKafka) ToKafkaConfigMap() *confluent.ConfigMap {
@@ -46,13 +46,10 @@ func dynamicKafkaClientConfigLoad() []byte {
 func publisherKafkaConfigLoader() {
 	viper.SetDefault("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "100000")
 	viper.SetDefault("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS", "1000")
-	viper.SetDefault("PUBLISHER_KAFKA_DELIVERY_REPORT_INTERVAL_MS", "60000")
-	viper.SetDefault("PUBLISHER_KAFKA_DELIVERY_REPORT_TOPIC_NAME", "clickstream-total-event-log")
+	viper.SetDefault("PUBLISHER_KAFKA_CLICKSTREAM_STATS_TOPIC_NAME", "clickstream-total-event-log")
 	viper.MergeConfig(bytes.NewBuffer(dynamicKafkaClientConfigLoad()))
-
 	PublisherKafka = publisherKafka{
-		FlushInterval:          util.MustGetInt("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS"),
-		DeliveryReportInterval: util.MustGetDuration("PUBLISHER_KAFKA_DELIVERY_REPORT_INTERVAL_MS", time.Millisecond),
-		DeliveryReportTopic:    util.MustGetString("PUBLISHER_KAFKA_DELIVERY_REPORT_TOPIC_NAME"),
+		FlushInterval:             util.MustGetInt("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS"),
+		ClickstreamStatsTopicName: util.MustGetString("PUBLISHER_KAFKA_CLICKSTREAM_STATS_TOPIC_NAME"),
 	}
 }

publisher/kafka.go

@@ -5,11 +5,8 @@ import (
 	"fmt"
 	"github.com/goto/raccoon/serialization"
 	"google.golang.org/protobuf/types/known/timestamppb"
-	"strings"
-	"sync/atomic"
-	"time"
-
 	"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
+	"strings"
 	// Importing librd to make it work on vendor mode
 	_ "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka/librdkafka"
 
@@ -24,8 +21,6 @@ const (
 	errLargeMessageSize = "Broker: Message size too large" //error msg while producing a message which is larger than message.max.bytes config
 )
 
-var DeliveryEventCount int64
-
 // KafkaProducer Produce data to kafka synchronously
 type KafkaProducer interface {
 	// ProduceBulk message to kafka. Block until all messages are sent. Return array of error. Order is not guaranteed.
@@ -38,31 +33,34 @@ func NewKafka() (*Kafka, error) {
 		return &Kafka{}, err
 	}
 	return &Kafka{
-		kp:                     kp,
-		flushInterval:          config.PublisherKafka.FlushInterval,
-		topicFormat:            config.EventDistribution.PublisherPattern,
-		deliveryReportInterval: config.PublisherKafka.DeliveryReportInterval,
-		deliveryReportTopic:    config.PublisherKafka.DeliveryReportTopic,
+		kp:            kp,
+		flushInterval: config.PublisherKafka.FlushInterval,
+		topicFormat:   config.EventDistribution.PublisherPattern,
+		clickstreamStats: ClickstreamStats{
+			config.PublisherKafka.ClickstreamStatsTopicName,
+		},
 	}, nil
 }
 
-func NewKafkaFromClient(client Client, flushInterval int, topicFormat string, deliveryReportInterval time.Duration,
-	topicName string) *Kafka {
+func NewKafkaFromClient(client Client, flushInterval int, topicFormat string,
+	clickstreamStats ClickstreamStats) *Kafka {
 	return &Kafka{
-		kp:                     client,
-		flushInterval:          flushInterval,
-		topicFormat:            topicFormat,
-		deliveryReportInterval: deliveryReportInterval,
-		deliveryReportTopic:    topicName,
+		kp:               client,
+		flushInterval:    flushInterval,
+		topicFormat:      topicFormat,
+		clickstreamStats: clickstreamStats,
 	}
 }
 
 type Kafka struct {
-	kp                     Client
-	flushInterval          int
-	topicFormat            string
-	deliveryReportInterval time.Duration
-	deliveryReportTopic    string
+	kp               Client
+	flushInterval    int
+	topicFormat      string
+	clickstreamStats ClickstreamStats
+}
+
+type ClickstreamStats struct {
+	topicName string
 }
 
 // ProduceBulk messages to kafka. Block until all messages are sent. Return array of error. Order of Errors is guaranteed.
@@ -102,7 +100,7 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 		totalProcessed++
 	}
 	// Wait for deliveryChannel as many as processed
-	localSuccesses := int64(0)
+	totalEventCount := int64(0)
 	for i := 0; i < totalProcessed; i++ {
 		d := <-deliveryChannel
 		m := d.(*kafka.Message)
@@ -114,13 +112,17 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 			order := m.Opaque.(int)
 			errors[order] = m.TopicPartition.Error
 		} else {
-			localSuccesses++
+			totalEventCount++
 		}
 	}
-
-	// Single atomic update per batch
-	if localSuccesses > 0 {
-		atomic.AddInt64(&DeliveryEventCount, localSuccesses)
+	// send the total successful delivered event count to kafka topic
+	if totalEventCount > 0 {
+		//build kafka message for total message count
+		msg := &pb.TotalEventCountMessage{
+			EventTimestamp: timestamppb.Now(),
+			EventCount:     int32(totalEventCount),
+		}
+		pr.produceClickstreamStats(pr.clickstreamStats.topicName, msg)
 	}
 
 	if allNil(errors) {
@@ -129,7 +131,8 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 	return BulkError{Errors: errors}
 }
 
-func (pr *Kafka) produceTotalEventMessage(topicName string, event *pb.TotalEventCountMessage) error {
+// produceClickstreamStats : method to produce stats to a kafka topic
+func (pr *Kafka) produceClickstreamStats(topicName string, event *pb.TotalEventCountMessage) error {
 	value, err := serialization.SerializeProto(event)
 	if err != nil {
 		return fmt.Errorf("failed to serialize proto: %w", err)
@@ -173,25 +176,6 @@ func (pr *Kafka) ReportStats() {
 	}
 }
 
-func (pr *Kafka) ReportDeliveryEventCount() {
-	ticker := time.NewTicker(pr.deliveryReportInterval)
-	defer ticker.Stop()
-
-	for range ticker.C {
-		// read the value
-		eventCount := atomic.LoadInt64(&DeliveryEventCount)
-		//build kafka message
-		msg := &pb.TotalEventCountMessage{
-			EventTimestamp: timestamppb.Now(),
-			EventCount:     int32(eventCount),
-		}
-		//produce to kafka
-		pr.produceTotalEventMessage(pr.deliveryReportTopic, msg)
-		//reset the counter
-		atomic.StoreInt64(&DeliveryEventCount, 0)
-	}
-}
-
 // Close wait for outstanding messages to be delivered within given flush interval timeout.
 func (pr *Kafka) Close() int {
 	remaining := pr.kp.Flush(pr.flushInterval)