[core] async kafka writing with error reporting

Author: Michal Tichák

common/event/writer.go

@@ -55,22 +55,28 @@ func (*DummyWriter) Close()                                         {}
 
 type KafkaWriter struct {
 	*kafka.Writer
+	toWriteChan chan kafka.Message
 }
 
 func NewWriterWithTopic(topic topic.Topic) *KafkaWriter {
-	return &KafkaWriter{
+	writer := &KafkaWriter{
 		Writer: &kafka.Writer{
 			Addr:                   kafka.TCP(viper.GetStringSlice("kafkaEndpoints")...),
 			Topic:                  string(topic),
 			Balancer:               &kafka.Hash{},
 			AllowAutoTopicCreation: true,
 		},
+		toWriteChan: make(chan kafka.Message, 1000),
 	}
+
+	go writer.writingLoop()
+	return writer
 }
 
 func (w *KafkaWriter) Close() {
 	if w != nil {
-		w.Close()
+		close(w.toWriteChan)
+		w.Writer.Close()
 	}
 }
 
@@ -80,6 +86,16 @@ func (w *KafkaWriter) WriteEvent(e interface{}) {
 	}
 }
 
+// TODO: we can optimise this to write multiple message at once
+func (w *KafkaWriter) writingLoop() {
+	for message := range w.toWriteChan {
+		err := w.WriteMessages(context.Background(), message)
+		if err != nil {
+			log.Errorf("failed to write async kafka message: %w", err)
+		}
+	}
+}
+
 type HasEnvID interface {
 	GetEnvironmentId() string
 }
@@ -200,9 +216,7 @@ func (w *KafkaWriter) doWriteEvent(key []byte, e *pb.Event) error {
 		message.Key = key
 	}
 
-	err = w.WriteMessages(context.Background(), message)
-	if err != nil {
-		return fmt.Errorf("failed to write event: %w", err)
-	}
+	w.toWriteChan <- message
+
 	return nil
 }