- Add background service, setup the consumer and logging in Program.cs

- Consumer working successfully 🎉

Author: Ashish Khanal

Consumer/Consumer.csproj

@@ -11,6 +11,8 @@
     <ItemGroup>
       <PackageReference Include="Confluent.Kafka" Version="2.3.0" />
       <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
+      <PackageReference Include="Confluent.SchemaRegistry" Version="2.3.0" />
+      <PackageReference Include="Confluent.SchemaRegistry.Serdes.Json" Version="2.3.0" />
     </ItemGroup>
 
     <ItemGroup>

Consumer/Program.cs

@@ -1,5 +1,14 @@
-using Microsoft.Extensions.Configuration;
-using Confluent.Kafka;
+using Confluent.Kafka;
+using Confluent.Kafka.SyncOverAsync;
+using Confluent.SchemaRegistry.Serdes;
+using Consumer.Workers;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);
 
 //using var host = Host.CreateApplicationBuilder(args).Build();
 //var config = host.Services.GetRequiredService<IConfiguration>(); // This works like below EXCEPT the user secrets part.
@@ -9,6 +18,25 @@
   .AddUserSecrets<Program>()
   .Build();
 
-var consumerConfig = config.GetSection("KafkaConsumer").Get<ConsumerConfig>()!;
+// Populate consumer config
+builder.Services.Configure<ConsumerConfig>(config.GetSection("KafkaConsumer"));
+
+builder.Services.AddSingleton<IConsumer<string, Biometrics>>(sp =>
+{
+  var consumerConfig = sp.GetRequiredService<IOptions<ConsumerConfig>>();
+  return new ConsumerBuilder<string, Biometrics>(consumerConfig.Value)
+    .SetValueDeserializer(new JsonDeserializer<Biometrics>().AsSyncOverAsync())
+    .Build();
+});
+
+builder.Services.AddHostedService<HeartRateZoneWorker>();
+
+using IHost host = builder.Build();
+// https://learn.microsoft.com/en-us/dotnet/core/extensions/logging?tabs=command-line#create-logs-in-main
+var logger = host.Services.GetRequiredService<ILogger<Program>>();
+logger.LogInformation("Host created. Now running it.");
+
+await host.RunAsync();
 
-Console.ReadLine();
+public record Biometrics(Guid DeviceId, List<HeartRate> HeartRates);
+public record HeartRate(DateTime DateTime, int Value);

Consumer/Workers/HeartRateZoneWorker.cs

@@ -0,0 +1,34 @@
+using Confluent.Kafka;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Consumer.Workers;
+
+public class HeartRateZoneWorker(
+    ILogger<HeartRateZoneWorker> logger,
+    IConsumer<string, Biometrics> consumer) : BackgroundService
+{
+    private const string BiometricsImportedTopicName = "BiometricsImported";
+    
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        consumer.Subscribe(BiometricsImportedTopicName);
+        
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            var result = consumer.Consume(stoppingToken);
+            logger.LogInformation("Message Received: {0}", result.Message.Value);
+            await Task.CompletedTask;
+            
+            //await Task.Delay(1000, stoppingToken);
+        }
+        
+        consumer.Close();
+    }
+    
+    public override async Task StopAsync(CancellationToken stoppingToken)
+    {
+        logger.LogInformation("HeartRateZoneWorker is stopping.");
+        await base.StopAsync(stoppingToken);
+    }
+}

Consumer/appsettings.json

@@ -8,9 +8,10 @@
     "BootstrapServers": "pkc-4rn2p.canadacentral.azure.confluent.cloud:9092",
     "SecurityProtocol": "SaslSsl",
     "SaslMechanism": "PLAIN",
-    "ClientId": "my-dotnet-kafka",
-    "GroupId": "my-dotnet-kafka-group",
+    "ClientId": "dotnet-kafka-consumer",
+    "GroupId": "dotnet-kafka-consumer",
     "AutoOffsetReset": "earliest",
+    "EnableAutoCommit": true,
     "SaslUsername": "comes from user-secrets' secrets.json",
     "SaslPassword": "comes from user-secrets' secrets.json"
   },

Producer/Program.cs

@@ -52,7 +52,7 @@
         var result = await producer.ProduceAsync(biometricsImportedTopicName, message);
         producer.Flush();
         app.Logger.LogInformation("Accepted Biometrics");
-        return TypedResults.Accepted("", result.Value); // result is of type <string, Biometrics>, so you're just returning the accepted Biometrics
+        return TypedResults.Accepted("", result.Value); // result.Value is just message.Value
     })
     .WithName("RecordMeasurements")
     .WithOpenApi();

Producer/appsettings.json

@@ -12,10 +12,11 @@
     "SaslMechanism": "PLAIN",
     "SaslUsername": "comes from user-secrets' secrets.json",
     "SaslPassword": "comes from user-secrets' secrets.json",
-    "ClientId": "my-dotnet-kafka"
+    "ClientId": "dotnet-kafka-producer"
   },
   "KafkaSchemaRegistry": {
     "URL": "https://psrc-gq7pv.westus2.azure.confluent.cloud",
-    "BasicAuthCredentialsSource": "UserInfo"
+    "BasicAuthCredentialsSource": "UserInfo",
+    "BasicAuthUserInfo": "comes from user-secrets' secrets.json"
   }
 }