consumer group per consumer type #GCPActive (#317)

src/Altinn.Notifications.Integrations/Configuration/KafkaSettings.cs

@@ -107,4 +107,9 @@ public class AdminSettings
     /// The list of topics the admin client is responsible for ensuring that exist
     /// </summary>
     public List<string> TopicList { get; set; } = new List<string>();
+
+    /// <summary>
+    /// The retention time in days for the topics
+    /// </summary>
+    public int RetentionTime { get; set; } = 7;
 }

src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs

@@ -32,7 +32,7 @@ protected KafkaConsumerBase(
 
         var consumerConfig = new ConsumerConfig(config.ConsumerSettings)
         {
-            GroupId = settings.Value.Consumer.GroupId,
+            GroupId = $"{settings.Value.Consumer.GroupId}-{GetType().Name.ToLower()}",
             EnableAutoCommit = false,
             EnableAutoOffsetStore = false,
             AutoOffsetReset = AutoOffsetReset.Earliest,

src/Altinn.Notifications.Integrations/Kafka/Producers/KafkaProducer.cs

@@ -99,7 +99,8 @@ private void EnsureTopicsExist()
                         {
                             Name = topic,
                             NumPartitions = TopicSpecification.NumPartitions,
-                            ReplicationFactor = TopicSpecification.ReplicationFactor
+                            ReplicationFactor = TopicSpecification.ReplicationFactor,
+                            Configs = TopicSpecification.Configs
                         }
                     }).Wait();
                     _logger.LogInformation("// KafkaProducer // EnsureTopicsExists // Topic '{Topic}' created successfully.", topic);

src/Altinn.Notifications.Integrations/Kafka/SharedClientConfig.cs

@@ -74,8 +74,14 @@ public SharedClientConfig(Configuration.KafkaSettings settings)
             consumerConfig.SaslUsername = settings.Consumer.SaslUsername;
             consumerConfig.SaslPassword = settings.Consumer.SaslPassword;
 
+            string retentionTime = settings.Admin.RetentionTime < 0 ? "-1" : TimeSpan.FromDays(settings.Admin.RetentionTime).TotalMilliseconds.ToString();
             topicSpec.NumPartitions = 6;
             topicSpec.ReplicationFactor = 3;
+            topicSpec.Configs = new Dictionary<string, string>()
+                                {
+                                    { "retention.ms", retentionTime },
+                                    { "cleanup.policy", "delete" }
+                                };
         }
 
         AdminClientSettings = adminConfig;