Introduction to Apache Kafka with .NET

public class KafkaProducer : IKafkaProducer
{
    private readonly IProducer<string, string> kafkaProducer;

    public KafkaProducer(string bootstrapServers)
    {
        var producerConfig = new ProducerConfig
        {
            BootstrapServers = bootstrapServers,
        };

        this.kafkaProducer = new ProducerBuilder<string, string>(producerConfig).Build();
    }

    public void SendMessage<T>(string topic, string key, T message)
    {
        var serialized_message = JsonConvert.SerializeObject(message);
        kafkaProducer.Produce(topic, new Message<string, string> { Key = key, Value = serialized_message });
    }

    public void Dispose()
    {
        kafkaProducer?.Dispose();
    }
}

IKafkaConsumer.cs

public interface IKafkaConsumer
{
    public void StartConsuming(string topic);
}

KafkaConsumer.cs: This is a Topic Consumer class, which is used to Read the Topic of Krom Kafka broker with the help of Confluent.Kafka package and with the method StartConsuming.

public class KafkaConsumer : IKafkaConsumer
{
    private readonly IConsumer<string, string> kafkaConsumer;

    public KafkaConsumer(string bootstrapServers, string groupId, string autoOffsetRest, string enableAutoOffsetStore)
    {
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = bootstrapServers,
            GroupId = groupId,
            AutoOffsetReset = ConvertToAutoOffsetReset(autoOffsetRest),// AutoOffsetReset.Earliest, 
            EnableAutoOffsetStore = Convert.ToBoolean(enableAutoOffsetStore),
        };

        this.kafkaConsumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
    }

    static AutoOffsetReset ConvertToAutoOffsetReset(string value)
    {
        return value.ToLower() switch
        {
            "earliest" => AutoOffsetReset.Earliest,
            "latest" => AutoOffsetReset.Latest,
            "error" => AutoOffsetReset.Error,
            _ => throw new ArgumentException($"Invalid AutoOffsetReset value: {value}")
        };
    }

    public void StartConsuming(string topic)
    {
        // Subscribe to the Kafka topic
        kafkaConsumer.Subscribe(topic);

        // Start an infinite loop to consume messages
        while (true)
        {
            try
            {
                // Consume messages from the Kafka topic
                var consumeResult = kafkaConsumer.Consume(CancellationToken.None);

                // Process the consumed message (implement your logic here)
                Console.WriteLine($"Received message: {consumeResult.Offset} \n");
                Console.WriteLine($"Received message: {consumeResult.Message.Key} \n") ;
                Console.WriteLine($"Received message: {consumeResult.Message.Value} \n\n\n");
            }
            catch (ConsumeException ex)
            {
                // Handle any errors that occur during message consumption
                Console.WriteLine($"Error consuming message: {ex.Error.Reason}");
            }
        }
    }

    public void Dispose()
    {
        // Properly dispose of the Kafka consumer
        kafkaConsumer?.Dispose();
    }
}

Step 11. Let's add 2 APIs for publishing and consuming the Topic to and from Kafka in the Program.cs file.

a) add the below code to read configuration related to Kafka from appsettings.json in Program.cs file.

var objBuilder = new ConfigurationBuilder()
          .SetBasePath(Directory.GetCurrentDirectory())
          .AddJsonFile("appSettings.json", optional: true, reloadOnChange: true);
IConfiguration configuration = objBuilder.Build();

//kafka
builder.Services.AddSingleton<IKafkaProducer>(provider =>
{
    var configuration = provider.GetRequiredService<IConfiguration>();
    var bootstrapServers = configuration["KafkaProducerConfig:bootstrapServers"];

    return new KafkaProducer(bootstrapServers);
});

builder.Services.AddSingleton<IKafkaConsumer>(provider =>
{
    var configuration = provider.GetRequiredService<IConfiguration>();
    var bootstrapServers = configuration["KafkaConsumerConfig:BootstrapServers"];
    var groupId = configuration["KafkaConsumerConfig:GroupId"];
    var autoOffsetRest = configuration["KafkaConsumerConfig:AutoOffsetRest"];
    var enableAutoOffsetStore = configuration["KafkaConsumerConfig:EnableAutoOffsetStore"];

    return new KafkaConsumer(bootstrapServers, groupId, autoOffsetRest, enableAutoOffsetStore);
});

b) Use existing weatherforecast API which got created along with the project to publish the Topic for Kafka.

In the below code, the Message is published with the Topic: message_weatherforecast with Key: Key_1.

app.MapGet("/weatherforecast", async([FromServices]IKafkaProducer kafkaProducer) =>
{
    var forecast = Enumerable.Range(1, 2).Select(index =>
        new WeatherForecast
        (
            DateTime.Now.AddDays(index),
            Random.Shared.Next(-20, 55),
            summaries[Random.Shared.Next(summaries.Length)]
        ))
        .ToArray();

    var topic = "message_weatherforecast";
    var key = "Key_1";
    kafkaProducer.SendMessage(topic, key, forecast);

    return forecast;
})
.WithName("GetWeatherForecast");

internal record WeatherForecast(DateTime Date, int TemperatureC, string? Summary)
{
    public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
}

c) Create a New API 'getForecastDetails' to read the Message that is published to Kafka.

In the below code, the Message is read with the Topic: message_weatherforecast.

app.MapGet("/getForecastDetails", async ([FromServices] IKafkaConsumer kafkaConsumer) =>
{    
    var topic = "message_weatherforecast";    

    kafkaConsumer.StartConsuming(topic);
})
.WithName("GetForecastDetails");

Step 12. Run the project and we should get below the swagger page.

Step 13. Now execute the 'WeatherForecast' API and check the response has appeared and that same has been published to Kafka via the Kafka UI tool with the topic name.

Verify Topic has been published in Kafka UI.

Verify the message present in the topic in the Data tab.

If the value is displayed in an encoded format, then modify Content Types to string at the Properties tab.

Step 14. Now try executing the 'GetForecastDetails' API and it will be running continuously to consume the message available in Kafka for the registered Topic.

The result of the above API can be viewed in the console. This will be the same as the data present in the Value column of the Kafka UI.

Key components and concepts of Kafka

1. Producer: Applications that publish streams of records into Kafka topics.
   Example: In the above example '/weatherforecast' API is a Producer that publishes streams of records.
2. Consumer: Applications that subscribe to topics and process the streams of records produced by the producers.
   Example: In the above example '/getForecastDetails' API is the Consumer for the streams of record that are published by the Producer.
3. Broker: Kafka runs as a cluster of one or more servers, each of which is called a broker.
   Example: In the above example the cluster running with port localhost:29092 is the Broker.
4. Topic: A category to which records are published. Topics in Kafka are always multi-subscriber.
   Example: In the above example Producer '/weatherforecast' API publishes streams of record into topic 'message_weatherforecast'.
5. Partition: Topics are divided into partitions, which allow parallel processing of data streams and scalability.
   Example: In the above example, a Partition can be identified in Kafka UI.
6. Offset: Each record within a partition has a unique sequential ID called an offset.
   Example: In the above example, Offset can be identified in Kafka UI.
7. Consumer Group: A set of consumers that jointly consume a set of subscribed topics. Kafka distributes the partitions of a topic among the members of the consumer group.
   Example: In the above example, the Consumer Group can be identified in Kafka UI.
8. ZooKeeper: Used for managing and coordinating Kafka brokers and consumers. However, recent versions of Kafka have been moving away from ZooKeeper for internal coordination.

Summary

Kafka is widely used in industries such as finance, telecommunications, retail, and more for various use cases, including log aggregation, data integration, real-time analytics, and event-driven microservices architectures.