feat(watermill): add watermill integration with observability support

- Add watermill client with Kafka and RabbitMQ backends
- Implement OTEL tracing and metrics middleware
- Add configurable retry middleware via config
- Fix logger adapter With() method to properly merge fields
- Add graceful shutdown with error collection
- Fix resource leak in subscriber error path
- Use explicit meter/tracer providers (DI pattern)
- Add metric descriptions and proper units (seconds)
- Fix context extraction in PublisherWrapper

Author: batazor

go.work

@@ -29,4 +29,5 @@ use (
 	./transactional_outbox
 	./types
 	./uow
+	./watermill
 )

go.work.sum

@@ -34,6 +34,7 @@ github.com/Djarvur/go-err113 v0.1.1/go.mod h1:IaWJdYFLg76t2ihfflPZnM1LIQszWOsFDh
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0=
 github.com/MirrexOne/unqueryvet v1.2.1/go.mod h1:IWwCwMQlSWjAIteW0t+28Q5vouyktfujzYznSIWiuOg=
 github.com/OpenPeeDeeP/depguard/v2 v2.2.1/go.mod h1:q4DKzC4UcVaAvcfd41CZh0PWpGgzrVxUYBlgKNGquUo=
+github.com/Shopify/toxiproxy/v2 v2.5.0/go.mod h1:yhM2epWtAmel9CB8r2+L+PCmhH6yH2pITaPAo7jxJl0=
 github.com/alecthomas/chroma/v2 v2.20.0/go.mod h1:e7tViK0xh/Nf4BYHl00ycY6rV7b8iXBksI9E359yNmA=
 github.com/alecthomas/go-check-sumtype v0.3.1/go.mod h1:A8TSiN3UPRw3laIgWEUOHHLPa6/r9MtoigdlP5h3K/E=
 github.com/alecthomas/kingpin/v2 v2.4.0/go.mod h1:0gyi0zQnjuFk8xrkNKamJoyUo382HRL7ATRpFZCw6tE=
@@ -254,6 +255,7 @@ github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFR
 github.com/sivchari/containedctx v1.0.3/go.mod h1:c1RDvCbnJLtH4lLcYD/GqwiBSSf4F5Qk0xld2rBqzJ4=
 github.com/snowflakedb/gosnowflake v1.6.19/go.mod h1:FM1+PWUdwB9udFDsXdfD58NONC0m+MlOSmQRvimobSM=
 github.com/sonatard/noctx v0.4.0/go.mod h1:64XdbzFb18XL4LporKXp8poqZtPKbCrqQ402CV+kJas=
+github.com/sony/gobreaker v1.0.0/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
 github.com/sourcegraph/go-diff v0.7.0/go.mod h1:iBszgVvyxdc8SFZ7gm69go2KDdt3ag071iBaWPF6cjs=
 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
 github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs=

mq/go.sum

@@ -46,6 +46,7 @@ github.com/ClickHouse/clickhouse-go v1.3.12/go.mod h1:EaI/sW7Azgz9UATzd5ZdZHRUhH
 github.com/IBM/sarama v1.46.3 h1:njRsX6jNlnR+ClJ8XmkO+CM4unbrNr/2vB5KK6UA+IE=
 github.com/IBM/sarama v1.46.3/go.mod h1:GTUYiF9DMOZVe3FwyGT+dtSPceGFIgA+sPc5u6CBwko=
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
+github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
 github.com/Microsoft/go-winio v0.4.15-0.20190919025122-fc70bd9a86b5/go.mod h1:tTuCMEN+UleMWgg9dVx4Hu52b1bJo+59jBh3ajtinzw=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
 github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=

watermill/backends/kafka/context.go

@@ -0,0 +1,56 @@
+package kafka
+
+import (
+	"context"
+	"time"
+)
+
+type contextKey int
+
+const (
+	_ contextKey = iota
+	partitionContextKey
+	partitionOffsetContextKey
+	timestampContextKey
+	keyContextKey
+)
+
+func setPartitionToCtx(ctx context.Context, partition int32) context.Context {
+	return context.WithValue(ctx, partitionContextKey, partition)
+}
+
+// MessagePartitionFromCtx returns Kafka partition of the consumed message
+func MessagePartitionFromCtx(ctx context.Context) (int32, bool) {
+	partition, ok := ctx.Value(partitionContextKey).(int32)
+	return partition, ok
+}
+
+func setPartitionOffsetToCtx(ctx context.Context, offset int64) context.Context {
+	return context.WithValue(ctx, partitionOffsetContextKey, offset)
+}
+
+// MessagePartitionOffsetFromCtx returns Kafka partition offset of the consumed message
+func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool) {
+	offset, ok := ctx.Value(partitionOffsetContextKey).(int64)
+	return offset, ok
+}
+
+func setMessageTimestampToCtx(ctx context.Context, timestamp time.Time) context.Context {
+	return context.WithValue(ctx, timestampContextKey, timestamp)
+}
+
+// MessageTimestampFromCtx returns Kafka internal timestamp of the consumed message
+func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool) {
+	timestamp, ok := ctx.Value(timestampContextKey).(time.Time)
+	return timestamp, ok
+}
+
+func setMessageKeyToCtx(ctx context.Context, key []byte) context.Context {
+	return context.WithValue(ctx, keyContextKey, key)
+}
+
+// MessageKeyFromCtx returns Kafka internal key of the consumed message
+func MessageKeyFromCtx(ctx context.Context) ([]byte, bool) {
+	key, ok := ctx.Value(keyContextKey).([]byte)
+	return key, ok
+}

watermill/backends/kafka/marshaler.go

@@ -0,0 +1,94 @@
+package kafka
+
+import (
+	"github.com/IBM/sarama"
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/pkg/errors"
+)
+
+const UUIDHeaderKey = "_watermill_message_uuid"
+
+// Marshaler marshals Watermill's message to Kafka message.
+type Marshaler interface {
+	Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
+}
+
+// Unmarshaler unmarshals Kafka's message to Watermill's message.
+type Unmarshaler interface {
+	Unmarshal(*sarama.ConsumerMessage) (*message.Message, error)
+}
+
+type MarshalerUnmarshaler interface {
+	Marshaler
+	Unmarshaler
+}
+
+type DefaultMarshaler struct{}
+
+func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error) {
+	if value := msg.Metadata.Get(UUIDHeaderKey); value != "" {
+		return nil, errors.Errorf("metadata %s is reserved by watermill for message UUID", UUIDHeaderKey)
+	}
+
+	headers := []sarama.RecordHeader{{
+		Key:   []byte(UUIDHeaderKey),
+		Value: []byte(msg.UUID),
+	}}
+	for key, value := range msg.Metadata {
+		headers = append(headers, sarama.RecordHeader{
+			Key:   []byte(key),
+			Value: []byte(value),
+		})
+	}
+
+	return &sarama.ProducerMessage{
+		Topic:   topic,
+		Value:   sarama.ByteEncoder(msg.Payload),
+		Headers: headers,
+	}, nil
+}
+
+func (DefaultMarshaler) Unmarshal(kafkaMsg *sarama.ConsumerMessage) (*message.Message, error) {
+	var messageID string
+	metadata := make(message.Metadata, len(kafkaMsg.Headers))
+
+	for _, header := range kafkaMsg.Headers {
+		if string(header.Key) == UUIDHeaderKey {
+			messageID = string(header.Value)
+		} else {
+			metadata.Set(string(header.Key), string(header.Value))
+		}
+	}
+
+	msg := message.NewMessage(messageID, kafkaMsg.Value)
+	msg.Metadata = metadata
+
+	return msg, nil
+}
+
+type GeneratePartitionKey func(topic string, msg *message.Message) (string, error)
+
+type kafkaJsonWithPartitioning struct {
+	DefaultMarshaler
+
+	generatePartitionKey GeneratePartitionKey
+}
+
+func NewWithPartitioningMarshaler(generatePartitionKey GeneratePartitionKey) MarshalerUnmarshaler {
+	return kafkaJsonWithPartitioning{generatePartitionKey: generatePartitionKey}
+}
+
+func (j kafkaJsonWithPartitioning) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error) {
+	kafkaMsg, err := j.DefaultMarshaler.Marshal(topic, msg)
+	if err != nil {
+		return nil, err
+	}
+
+	key, err := j.generatePartitionKey(topic, msg)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot generate partition key")
+	}
+	kafkaMsg.Key = sarama.ByteEncoder(key)
+
+	return kafkaMsg, nil
+}

watermill/backends/kafka/marshaler_test.go

@@ -0,0 +1,118 @@
+package kafka_test
+
+import (
+	"testing"
+
+	"github.com/IBM/sarama"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+func TestDefaultMarshaler_MarshalUnmarshal(t *testing.T) {
+	m := kafka.DefaultMarshaler{}
+
+	msg := message.NewMessage(watermill.NewUUID(), []byte("payload"))
+	msg.Metadata.Set("foo", "bar")
+
+	marshaled, err := m.Marshal("topic", msg)
+	require.NoError(t, err)
+
+	unmarshaledMsg, err := m.Unmarshal(producerToConsumerMessage(marshaled))
+	require.NoError(t, err)
+
+	assert.True(t, msg.Equals(unmarshaledMsg))
+}
+
+func BenchmarkDefaultMarshaler_Marshal(b *testing.B) {
+	m := kafka.DefaultMarshaler{}
+
+	msg := message.NewMessage(watermill.NewUUID(), []byte("payload"))
+	msg.Metadata.Set("foo", "bar")
+
+	for i := 0; i < b.N; i++ {
+		_, _ = m.Marshal("foo", msg)
+	}
+}
+
+func BenchmarkDefaultMarshaler_Unmarshal(b *testing.B) {
+	m := kafka.DefaultMarshaler{}
+
+	msg := message.NewMessage(watermill.NewUUID(), []byte("payload"))
+	msg.Metadata.Set("foo", "bar")
+
+	marshaled, err := m.Marshal("foo", msg)
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	consumedMsg := producerToConsumerMessage(marshaled)
+
+	for i := 0; i < b.N; i++ {
+		_, _ = m.Unmarshal(consumedMsg)
+	}
+}
+
+func TestWithPartitioningMarshaler_MarshalUnmarshal(t *testing.T) {
+	m := kafka.NewWithPartitioningMarshaler(func(topic string, msg *message.Message) (string, error) {
+		return msg.Metadata.Get("partition"), nil
+	})
+
+	partitionKey := "1"
+	msg := message.NewMessage(watermill.NewUUID(), []byte("payload"))
+	msg.Metadata.Set("partition", partitionKey)
+
+	producerMsg, err := m.Marshal("topic", msg)
+	require.NoError(t, err)
+
+	unmarshaledMsg, err := m.Unmarshal(producerToConsumerMessage(producerMsg))
+	require.NoError(t, err)
+
+	assert.True(t, msg.Equals(unmarshaledMsg))
+
+	assert.NoError(t, err)
+
+	producerKey, err := producerMsg.Key.Encode()
+	require.NoError(t, err)
+
+	assert.Equal(t, string(producerKey), partitionKey)
+}
+
+func producerToConsumerMessage(producerMessage *sarama.ProducerMessage) *sarama.ConsumerMessage {
+	var key []byte
+
+	if producerMessage.Key != nil {
+		var err error
+		key, err = producerMessage.Key.Encode()
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	var value []byte
+	if producerMessage.Value != nil {
+		var err error
+		value, err = producerMessage.Value.Encode()
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	var headers []*sarama.RecordHeader
+	for i := range producerMessage.Headers {
+		headers = append(headers, &producerMessage.Headers[i])
+	}
+
+	return &sarama.ConsumerMessage{
+		Key:       key,
+		Value:     value,
+		Topic:     producerMessage.Topic,
+		Partition: producerMessage.Partition,
+		Offset:    producerMessage.Offset,
+		Timestamp: producerMessage.Timestamp,
+		Headers:   headers,
+	}
+}