feat(duely): kafka sasl auth

Duely/docker-compose.yml

@@ -65,6 +65,7 @@ services:
       Kafka__TaskiTopic: taski.testing
       Kafka__ExeshTopic: exesh.step-updates
       Kafka__BootstrapServers: kafka:9092
+      Kafka__SaslAuth: false
     depends_on:
       postgres:
         condition: service_healthy

Duely/src/Duely.Infrastructure.MessageBus.Kafka/ExeshSubmissionStatusConsumer.cs

@@ -30,6 +30,14 @@ public ExeshSubmissionStatusConsumer(IServiceScopeFactory scopeFactory, IOptions
             GroupId = kafkaOptions.Value.GroupId,
             AutoOffsetReset = AutoOffsetReset.Earliest
         };
+
+        if (kafkaOptions.Value.SaslAuth)
+        {
+            config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
+            config.SaslMechanism = SaslMechanism.ScramSha512;
+            config.SaslUsername = kafkaOptions.Value.SaslUsername;
+            config.SaslPassword = kafkaOptions.Value.SaslPassword;
+        }
 
         _consumer = new ConsumerBuilder<string, ExeshStatusEvent>(config)
             .SetValueDeserializer(new KafkaValueDeserializer<ExeshStatusEvent>())

Duely/src/Duely.Infrastructure.MessageBus.Kafka/KafkaOptions.cs

@@ -8,4 +8,7 @@ public sealed class KafkaOptions
     public required string TaskiTopic { get; init; }
     public required string ExeshTopic { get; init; }
     public required string GroupId { get; init; }
+    public required bool SaslAuth { get; init; }
+    public string? SaslUsername { get; init; }
+    public string? SaslPassword { get; init; }
 }

Duely/src/Duely.Infrastructure.MessageBus.Kafka/TaskiSubmissionStatusConsumer.cs

@@ -28,6 +28,14 @@ public TaskiSubmissionStatusConsumer(IServiceScopeFactory scopeFactory, IOptions
             BootstrapServers = kafkaOptions.Value.BootstrapServers,
             GroupId = kafkaOptions.Value.GroupId
         };
+
+        if (kafkaOptions.Value.SaslAuth)
+        {
+            config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
+            config.SaslMechanism = SaslMechanism.ScramSha512;
+            config.SaslUsername = kafkaOptions.Value.SaslUsername;
+            config.SaslPassword = kafkaOptions.Value.SaslPassword;
+        }
 
         _consumer = new ConsumerBuilder<string, TaskiStatusEvent>(config)
             .SetValueDeserializer(new KafkaValueDeserializer<TaskiStatusEvent>())

Duely/src/Duely/appsettings.Development.json

@@ -18,6 +18,7 @@
     "BootstrapServers": "localhost:29092",
     "TaskiTopic": "taski.testing",
     "ExeshTopic": "exesh.step-updates",
-    "GroupId": "duely"
+    "GroupId": "duely",
+    "SaslAuth": false
   }
 }