fire and forget kafka messages for event notification
@@ -4,6 +4,7 @@ import (
"context"
"crypto/tls"
"encoding/json"
+ "time"
"github.com/kelseyhightower/envconfig"
kafka "github.com/segmentio/kafka-go"
@@ -11,6 +12,10 @@ import (
"github.com/segmentio/kafka-go/sasl/plain"
)
+const (
+ WriteTimeout time.Duration = 5 * time.Second
+)
+
//go:generate mockgen -source=event_stream.go -package=stream -destination=mock_event_stream.go
type KafkaProducer interface {
@@ -48,6 +53,8 @@ func newProducer(config *KafkaConfig) KafkaProducer {
Topic: config.Topic,
Balancer: &kafka.ReferenceHash{},
Compression: compress.Gzip,
+ Async: true,
+ WriteTimeout: WriteTimeout,
}
if config.ClientID != "" && config.ClientSecret != "" {
mechanism := &plain.Mechanism{
@@ -100,6 +107,6 @@ func (w *KafkaWriter) Write(ctx context.Context, eventName string, key []byte, p
Key: key,
Value: value,
}
-
+ // If Async is true, this will always return nil
return w.kafkaProducer.WriteMessages(ctx, msg)
}
