feat: add support for setting the retention of dlq topics

openmeterio · Aug 15, 2024 · 8831608 · 8831608
1 parent 0b07819
commit 8831608
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 0 deletions.
diff --git a/cmd/balance-worker/main.go b/cmd/balance-worker/main.go
@@ -341,6 +341,7 @@ func initEventPublisherDriver(ctx context.Context, broker watermillkafka.BrokerO
 		provisionTopics = append(provisionTopics, watermillkafka.AutoProvisionTopic{
 			Topic:         conf.BalanceWorker.DLQ.Topic,
 			NumPartitions: int32(conf.BalanceWorker.DLQ.AutoProvision.Partitions),
+			Retention:     conf.BalanceWorker.DLQ.AutoProvision.Retention,
 		})
 	}
 

diff --git a/cmd/notification-service/main.go b/cmd/notification-service/main.go
@@ -334,6 +334,7 @@ func initEventPublisherDriver(ctx context.Context, logger *slog.Logger, conf con
 		provisionTopics = append(provisionTopics, watermillkafka.AutoProvisionTopic{
 			Topic:         conf.Notification.Consumer.DLQ.Topic,
 			NumPartitions: int32(conf.Notification.Consumer.DLQ.AutoProvision.Partitions),
+			Retention:     conf.BalanceWorker.DLQ.AutoProvision.Retention,
 		})
 	}
 

diff --git a/internal/watermill/driver/kafka/topic_provision.go b/internal/watermill/driver/kafka/topic_provision.go
@@ -3,13 +3,16 @@ package kafka
 import (
 	"context"
 	"log/slog"
+	"strconv"
+	"time"
 
 	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
 )
 
 type AutoProvisionTopic struct {
 	Topic         string
 	NumPartitions int32
+	Retention     time.Duration
 }
 
 // provisionTopics creates the topics if they don't exist. This relies on the confluent kafka lib, as the sarama doesn't seem to
@@ -26,10 +29,15 @@ func provisionTopics(ctx context.Context, logger *slog.Logger, config kafka.Conf
 	defer adminClient.Close()
 
 	for _, topic := range topics {
+		topicConfig := map[string]string{}
+		if topic.Retention > 0 {
+			topicConfig["retention.ms"] = strconv.FormatInt(topic.Retention.Milliseconds(), 10)
+		}
 		result, err := adminClient.CreateTopics(ctx, []kafka.TopicSpecification{
 			{
 				Topic:         topic.Topic,
 				NumPartitions: int(topic.NumPartitions),
+				Config:        topicConfig,
 			},
 		})
 		if err != nil {