feat: add support for producing kakfa messages asynchronously

Author: doriandekoning

exporter/kafkaexporter/exporter.go

@@ -19,10 +19,28 @@ type MessageSender interface {
 	SendMessages(msgs []*sarama.ProducerMessage) error
 }
 
+type AsyncMessageSender struct {
+	sarama.AsyncProducer
+	logger *fflog.FFLogger
+}
+
+func (a *AsyncMessageSender) SendMessages(msgs []*sarama.ProducerMessage) error {
+	for len(msgs) > 0 {
+		select {
+		case err := <-a.AsyncProducer.Errors():
+			a.logger.Warn("Failed to produce message: %w", err)
+		case a.AsyncProducer.Input() <- msgs[0]:
+			msgs = msgs[1:]
+		}
+	}
+	return nil
+}
+
 // Settings contains Kafka-specific configurations needed for message creation
 type Settings struct {
 	Topic     string   `json:"topic"`
 	Addresses []string `json:"addresses"`
+	Async     bool     `json:"async"`
 	*sarama.Config
 }
 
@@ -41,18 +59,24 @@ type Exporter struct {
 	// dialer will create the producer. This field is added for dependency injection during testing as sarama
 	// has the annoying tendency to dial as soon as a producer is created.
 	dialer func(addrs []string, config *sarama.Config) (MessageSender, error)
+
+	logger *fflog.FFLogger
 }
 
 // Export will produce a message to the Kafka topic. The message's value will contain the event encoded in the
 // selected format. Messages are published synchronously and will error immediately on failure.
-func (e *Exporter) Export(_ context.Context, _ *fflog.FFLogger, featureEvents []exporter.FeatureEvent) error {
+func (e *Exporter) Export(_ context.Context, logger *fflog.FFLogger, featureEvents []exporter.FeatureEvent) error {
 	if e.sender == nil {
 		err := e.initializeProducer()
 		if err != nil {
 			return fmt.Errorf("writer: %w", err)
 		}
 	}
 
+	if e.logger == nil {
+		e.logger = logger
+	}
+
 	messages := make([]*sarama.ProducerMessage, 0, len(featureEvents))
 	for _, event := range featureEvents {
 		data, err := e.formatMessage(event)
@@ -90,6 +114,13 @@ func (e *Exporter) initializeProducer() error {
 	if e.dialer == nil {
 		e.dialer = func(addrs []string, config *sarama.Config) (MessageSender, error) {
 			// Adapter for the function to comply with the MessageSender interface return
+			if e.Settings.Async {
+				asyncProducer, err := sarama.NewAsyncProducer(addrs, config)
+				if err != nil {
+					return nil, err
+				}
+				return &AsyncMessageSender{AsyncProducer: asyncProducer, logger: e.logger}, nil //TODO Close should be called on shutdown
+			}
 			return sarama.NewSyncProducer(addrs, config)
 		}
 	}