Merge pull request #1309 from openmeterio/fix/use-confluent-kafka-client

fix: confluent cloud connectivity issues

cmd/balance-worker/main.go

@@ -230,7 +230,7 @@ func main() {
 	}
 
 	// Create publisher
-	publishers, err := initEventPublisher(logger, conf, metricMeter)
+	publishers, err := initEventPublisher(ctx, logger, conf, metricMeter)
 	if err != nil {
 		logger.Error("failed to initialize event publisher", slog.String("error", err.Error()))
 		os.Exit(1)
@@ -359,7 +359,7 @@ type eventPublishers struct {
 	eventPublisher     publisher.Publisher
 }
 
-func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*eventPublishers, error) {
+func initEventPublisher(ctx context.Context, logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*eventPublishers, error) {
 	provisionTopics := []watermillkafka.AutoProvisionTopic{}
 	if conf.BalanceWorker.DLQ.AutoProvision.Enabled {
 		provisionTopics = append(provisionTopics, watermillkafka.AutoProvisionTopic{
@@ -368,12 +368,13 @@ func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMe
 		})
 	}
 
-	eventDriver, err := watermillkafka.NewPublisher(watermillkafka.PublisherOptions{
+	eventDriver, err := watermillkafka.NewPublisher(ctx, watermillkafka.PublisherOptions{
 		KafkaConfig:     conf.Ingest.Kafka.KafkaConfiguration,
 		ProvisionTopics: provisionTopics,
 		ClientID:        otelName,
 		Logger:          logger,
 		MetricMeter:     metricMeter,
+		DebugLogging:    conf.Telemetry.Log.Level == slog.LevelDebug,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create event driver: %w", err)

cmd/notification-service/main.go

@@ -229,7 +229,7 @@ func main() {
 	}
 
 	// Create publisher
-	publishers, err := initEventPublisher(logger, conf, metricMeter)
+	publishers, err := initEventPublisher(ctx, logger, conf, metricMeter)
 	if err != nil {
 		logger.Error("failed to initialize event publisher", slog.String("error", err.Error()))
 		os.Exit(1)
@@ -354,7 +354,7 @@ type eventPublishers struct {
 	eventPublisher     publisher.Publisher
 }
 
-func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*eventPublishers, error) {
+func initEventPublisher(ctx context.Context, logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*eventPublishers, error) {
 	provisionTopics := []watermillkafka.AutoProvisionTopic{}
 	if conf.NotificationService.Consumer.DLQ.AutoProvision.Enabled {
 		provisionTopics = append(provisionTopics, watermillkafka.AutoProvisionTopic{
@@ -363,12 +363,13 @@ func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMe
 		})
 	}
 
-	eventDriver, err := watermillkafka.NewPublisher(watermillkafka.PublisherOptions{
+	eventDriver, err := watermillkafka.NewPublisher(ctx, watermillkafka.PublisherOptions{
 		KafkaConfig:     conf.Ingest.Kafka.KafkaConfiguration,
 		ProvisionTopics: provisionTopics,
 		ClientID:        otelName,
 		Logger:          logger,
 		MetricMeter:     metricMeter,
+		DebugLogging:    conf.Telemetry.Log.Level == slog.LevelDebug,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create event driver: %w", err)

cmd/server/main.go

@@ -213,7 +213,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	eventPublishers, err := initEventPublisher(logger, conf, metricMeter)
+	eventPublishers, err := initEventPublisher(ctx, logger, conf, metricMeter)
 	if err != nil {
 		logger.Error("failed to initialize event publisher", "error", err)
 		os.Exit(1)
@@ -435,7 +435,7 @@ type publishers struct {
 	driver         message.Publisher
 }
 
-func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*publishers, error) {
+func initEventPublisher(ctx context.Context, logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (*publishers, error) {
 	if !conf.Events.Enabled {
 		publisher, err := publisher.NewPublisher(publisher.PublisherOptions{
 			Publisher: &noop.Publisher{},
@@ -458,12 +458,13 @@ func initEventPublisher(logger *slog.Logger, conf config.Configuration, metricMe
 		})
 	}
 
-	eventDriver, err := watermillkafka.NewPublisher(watermillkafka.PublisherOptions{
+	eventDriver, err := watermillkafka.NewPublisher(ctx, watermillkafka.PublisherOptions{
 		KafkaConfig:     conf.Ingest.Kafka.KafkaConfiguration,
 		ProvisionTopics: provisionTopics,
 		ClientID:        otelName,
 		Logger:          logger,
 		MetricMeter:     metricMeter,
+		DebugLogging:    conf.Telemetry.Log.Level == slog.LevelDebug,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create event driver: %w", err)

cmd/sink-worker/main.go

@@ -161,7 +161,7 @@ func main() {
 	var group run.Group
 
 	// initialize system event producer
-	ingestEventFlushHandler, err := initIngestEventPublisher(logger, conf, metricMeter)
+	ingestEventFlushHandler, err := initIngestEventPublisher(ctx, logger, conf, metricMeter)
 	if err != nil {
 		logger.Error("failed to initialize event publisher", "error", err)
 		os.Exit(1)
@@ -220,15 +220,16 @@ func main() {
 	}
 }
 
-func initIngestEventPublisher(logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (flushhandler.FlushEventHandler, error) {
+func initIngestEventPublisher(ctx context.Context, logger *slog.Logger, conf config.Configuration, metricMeter metric.Meter) (flushhandler.FlushEventHandler, error) {
 	if !conf.Events.Enabled {
 		return nil, nil
 	}
 
-	eventDriver, err := watermillkafka.NewPublisher(watermillkafka.PublisherOptions{
-		KafkaConfig: conf.Ingest.Kafka.KafkaConfiguration,
-		ClientID:    otelName,
-		Logger:      logger,
+	eventDriver, err := watermillkafka.NewPublisher(ctx, watermillkafka.PublisherOptions{
+		KafkaConfig:  conf.Ingest.Kafka.KafkaConfiguration,
+		ClientID:     otelName,
+		Logger:       logger,
+		DebugLogging: conf.Telemetry.Log.Level == slog.LevelDebug,
 
 		ProvisionTopics: []watermillkafka.AutoProvisionTopic{
 			{

internal/watermill/driver/kafka/logger.go

@@ -0,0 +1,23 @@
+package kafka
+
+import (
+	"fmt"
+)
+
+type LoggerFunc func(fmt string, args ...any)
+
+type SaramaLoggerAdaptor struct {
+	loggerFunc LoggerFunc
+}
+
+func (s *SaramaLoggerAdaptor) Print(v ...interface{}) {
+	s.loggerFunc(fmt.Sprint(v...))
+}
+
+func (s *SaramaLoggerAdaptor) Printf(format string, v ...interface{}) {
+	s.loggerFunc(fmt.Sprintf(format, v...))
+}
+
+func (s *SaramaLoggerAdaptor) Println(v ...interface{}) {
+	s.loggerFunc(fmt.Sprint(v...))
+}

internal/watermill/driver/kafka/publisher.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -26,6 +27,7 @@ type PublisherOptions struct {
 	Logger          *slog.Logger
 	MetricMeter     otelmetric.Meter
 	MeterPrefix     string
+	DebugLogging    bool
 }
 
 func (o *PublisherOptions) Validate() error {
@@ -47,7 +49,7 @@ func (o *PublisherOptions) Validate() error {
 	return nil
 }
 
-func NewPublisher(in PublisherOptions) (*kafka.Publisher, error) {
+func NewPublisher(ctx context.Context, in PublisherOptions) (*kafka.Publisher, error) {
 	if err := in.Validate(); err != nil {
 		return nil, err
 	}
@@ -66,18 +68,35 @@ func NewPublisher(in PublisherOptions) (*kafka.Publisher, error) {
 	wmConfig.OverwriteSaramaConfig.Metadata.RefreshFrequency = in.KafkaConfig.TopicMetadataRefreshInterval.Duration()
 	wmConfig.OverwriteSaramaConfig.ClientID = "openmeter/balance-worker"
 
-	switch in.KafkaConfig.SecurityProtocol {
-	case "SASL_SSL":
+	// These are globals, so we cannot append the publisher/subscriber name to them
+	sarama.Logger = &SaramaLoggerAdaptor{
+		loggerFunc: in.Logger.Info,
+	}
+
+	if in.DebugLogging {
+		sarama.DebugLogger = &SaramaLoggerAdaptor{
+			loggerFunc: in.Logger.Debug,
+		}
+	}
+
+	if in.KafkaConfig.SecurityProtocol == "SASL_SSL" {
 		wmConfig.OverwriteSaramaConfig.Net.SASL.Enable = true
-		wmConfig.OverwriteSaramaConfig.Net.SASL.User = in.KafkaConfig.SaslUsername
-		wmConfig.OverwriteSaramaConfig.Net.SASL.Password = in.KafkaConfig.SaslPassword
-		wmConfig.OverwriteSaramaConfig.Net.SASL.Mechanism = sarama.SASLMechanism(in.KafkaConfig.SecurityProtocol)
+		wmConfig.OverwriteSaramaConfig.Net.SASL.Handshake = true
+
 		wmConfig.OverwriteSaramaConfig.Net.TLS.Enable = true
 		wmConfig.OverwriteSaramaConfig.Net.TLS.Config = &tls.Config{}
-	default:
+
+		switch in.KafkaConfig.SaslMechanisms {
+		case "PLAIN":
+			wmConfig.OverwriteSaramaConfig.Net.SASL.User = in.KafkaConfig.SaslUsername
+			wmConfig.OverwriteSaramaConfig.Net.SASL.Password = in.KafkaConfig.SaslPassword
+			wmConfig.OverwriteSaramaConfig.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+		default:
+			return nil, fmt.Errorf("unsupported SASL mechanism: %s", in.KafkaConfig.SaslMechanisms)
+		}
 	}
 
-	// Producer specific settings
+	wmConfig.OverwriteSaramaConfig.Producer.Retry.Max = 10
 	wmConfig.OverwriteSaramaConfig.Producer.Return.Successes = true
 
 	meterRegistry, err := metrics.NewRegistry(metrics.NewRegistryOptions{
@@ -95,7 +114,7 @@ func NewPublisher(in PublisherOptions) (*kafka.Publisher, error) {
 		return nil, err
 	}
 
-	if err := provisionTopics(in.KafkaConfig.Broker, wmConfig.OverwriteSaramaConfig, in.ProvisionTopics); err != nil {
+	if err := provisionTopics(ctx, in.Logger, in.KafkaConfig.CreateKafkaConfig(), in.ProvisionTopics); err != nil {
 		return nil, err
 	}
 

internal/watermill/driver/kafka/topic_provision.go

@@ -1,35 +1,50 @@
 package kafka
 
 import (
-	"github.com/IBM/sarama"
+	"context"
+	"log/slog"
 
-	"github.com/openmeterio/openmeter/pkg/errorsx"
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
 )
 
 type AutoProvisionTopic struct {
 	Topic         string
 	NumPartitions int32
 }
 
-func provisionTopics(broker string, config *sarama.Config, topics []AutoProvisionTopic) error {
-	admin, err := sarama.NewClusterAdmin([]string{broker}, config)
+// provisionTopics creates the topics if they don't exist. This relies on the confluent kafka lib, as the sarama doesn't seem to
+// properly support interacting with the confluent cloud.
+func provisionTopics(ctx context.Context, logger *slog.Logger, config kafka.ConfigMap, topics []AutoProvisionTopic) error {
+	// This is not supported on admin client, so we need to remove it
+	delete(config, "go.logs.channel.enable")
+
+	adminClient, err := kafka.NewAdminClient(&config)
 	if err != nil {
 		return err
 	}
-	defer admin.Close()
+
+	defer adminClient.Close()
 
 	for _, topic := range topics {
-		err := admin.CreateTopic(topic.Topic, &sarama.TopicDetail{
-			NumPartitions:     topic.NumPartitions,
-			ReplicationFactor: -1, // use default
-		}, false)
+		result, err := adminClient.CreateTopics(ctx, []kafka.TopicSpecification{
+			{
+				Topic:         topic.Topic,
+				NumPartitions: int(topic.NumPartitions),
+			},
+		})
 		if err != nil {
-			if topicError, ok := errorsx.ErrorAs[*sarama.TopicError](err); ok && topicError.Err == sarama.ErrTopicAlreadyExists {
-				continue
-			}
-
 			return err
 		}
+
+		for _, r := range result {
+			code := r.Error.Code()
+
+			if code == kafka.ErrTopicAlreadyExists {
+				logger.Debug("topic already exists", slog.String("topic", topic.Topic))
+			} else if code != kafka.ErrNoError {
+				return r.Error
+			}
+		}
 	}
 
 	return nil

openmeter/watermill/driver/kafka/driver.go

@@ -1,6 +1,8 @@
 package kafka
 
 import (
+	"context"
+
 	"github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/cloudevents/sdk-go/v2/event"
@@ -17,8 +19,8 @@ type (
 	AutoProvisionTopic = watermillkafka.AutoProvisionTopic
 )
 
-func NewPublisher(in PublisherOptions) (*kafka.Publisher, error) {
-	return watermillkafka.NewPublisher(in)
+func NewPublisher(ctx context.Context, in PublisherOptions) (*kafka.Publisher, error) {
+	return watermillkafka.NewPublisher(ctx, in)
 }
 
 func AddPartitionKeyFromSubject(watermillIn *message.Message, cloudEvent event.Event) (*message.Message, error) {