KafkaToolkit.Net 1.0.0

Kafka Client Configuration

This document shows how to configure generic producers and consumers in the Kafka client package without tying them to any specific message type or service.

1. appsettings.json

Configure the Kafka section in your appsettings.json:

{
  "Kafka": {
    "BootstrapServers": "localhost:9092",
    "GroupId": "my-service-group",
    "TopicName": "my-topic"
  }
}

2. Program.cs (Configuration & Service Registration)

Bind the configuration and register Kafka producers, consumers, and the hosted service in a generic way:

using Confluent.Kafka;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var builder = WebApplication.CreateBuilder(args);

// 2.1 Read generic Kafka settings
var kafkaConfig = builder.Configuration.GetSection("Kafka");
var bootstrapServers = kafkaConfig["BootstrapServers"];
var groupId          = kafkaConfig["GroupId"];
var topicName        = kafkaConfig["TopicName"];

// 2.2 Register a generic Kafka producer
builder.Services.AddKafkaProducer("DefaultProducer", _ => new ProducerConfig
{
    BootstrapServers = bootstrapServers
});

// 2.3 Register a generic Kafka consumer
builder.Services.AddKafkaConsumer("DefaultConsumer", _ => new ConsumerConfig
{
    BootstrapServers      = bootstrapServers,
    GroupId               = groupId,
    AutoOffsetReset       = AutoOffsetReset.Earliest,
    EnableAutoCommit      = true,
    EnablePartitionEof    = true
});

// 2.4 Register the generic background hosted service
builder.Services.AddHostedService<GenericConsumerHostedService>();

var app = builder.Build();
app.Run();

3. GenericConsumerHostedService

A BackgroundService that subscribes to the configured topic, consumes messages, deserializes them to a generic type, and hands them off to a pluggable handler:

using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;

public class GenericConsumerHostedService : BackgroundService
{
    private readonly IBaseConsumer         _consumer;
    private readonly string                _topic;
    private readonly ILogger<GenericConsumerHostedService> _logger;

    public GenericConsumerHostedService(
        IKafkaFactory factory,
        IConfiguration configuration,
        ILogger<GenericConsumerHostedService> logger)
    {
        _consumer = factory.GetConsumer("DefaultConsumer");
        _logger   = logger;
        _topic    = configuration["Kafka:TopicName"];
    }

    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        _consumer.Subscribe(_topic);
        _logger.LogInformation("Subscribed to topic {Topic}", _topic);

        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                var msg = await _consumer.ConsumeAsync(cancellationToken);
                if (msg?.Message?.Value != null)
                {
                    var payload = JsonConvert.DeserializeObject<dynamic>(msg.Message.Value);
                    _logger.LogInformation("Message processed");
                }
            }
            catch (OperationCanceledException) { /* graceful exit */ }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error consuming message");
            }
        }
    }
}

With this single Markdown file you have:

1. Configuration schema for appsettings.json.
2. Service registration in Program.cs for generic producers and consumers.
3. Implementation of a generic consumer hosted service that delegates processing to a handler.