feat: add support for producing kakfa messages asynchronously

Author: doriandekoning

exporter/kafkaexporter/exporter.go

@@ -19,10 +19,22 @@ type MessageSender interface {
 	SendMessages(msgs []*sarama.ProducerMessage) error
 }
 
+type AsyncMessageSender struct {
+	sarama.AsyncProducer
+}
+
+func (a *AsyncMessageSender) SendMessages(msgs []*sarama.ProducerMessage) error {
+	for _, msg := range msgs {
+		a.AsyncProducer.Input() <- msg
+	}
+	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
 }
 
@@ -90,6 +102,14 @@ 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
+				}
+				e.Settings.Config.Producer.Return.Errors = false //TODO these should be read
+				return &AsyncMessageSender{AsyncProducer: asyncProducer}, nil
+			}
 			return sarama.NewSyncProducer(addrs, config)
 		}
 	}