Kafka With .Net Core

var config = new ProducerConfig {
    BootstrapServers = "localhost:9092",
        ClientId = "my-app",
        BrokerAddressFamily = BrokerAddressFamily.V4,
};

After that, we will create a message and send it using the producer object created above.

using
var producer = new ProducerBuilder < Null,
    string > (config).Build();
Console.WriteLine("Please enter the message you want to send");
var input = Console.ReadLine();
var message = new Message < Null,
    string > {
        Value = input
    };
var deliveryReport = await producer.ProduceAsync("my-topic", message);

In the end, we are just printing the delivery report information.

So our entire method will look something like this

public async Task CreateMessage() {
    var config = new ProducerConfig {
        BootstrapServers = "localhost:9092",
            ClientId = "my-app",
            BrokerAddressFamily = BrokerAddressFamily.V4,
    };
    using
    var producer = new ProducerBuilder < Null,
        string > (config).Build();
    Console.WriteLine("Please enter the message you want to send");
    var input = Console.ReadLine();
    var message = new Message < Null,
        string > {
            Value = input
        };
    var deliveryReport = await producer.ProduceAsync("my-topic", message);
    Console.WriteLine($ "Message delivered to {deliveryReport.TopicPartitionOffset}");
}

After this, in the Program.cs, we will create the object of our class and call our CreateMessage method.

ProduceMessage produceMessage = new ProduceMessage();
produceMessage.CreateMessage().Wait();

Our work in the KafkaProducer has been done. Now we will create a "KafkaConsumer" project, another console application, to consume and read the message sent out by the topic mentioned in the above code.

Install the Confluent.Kafka NuGet package using the Package Manager Console in Visual Studio or the dotnet CLI:

In this newly created project, we will create a new file called ConsumeMessage.cs, and we will create a new method in this file called "ReadMessage".

We will create the object of ConsumerConfig and provide our required configurations.

var config = new ConsumerConfig {
    BootstrapServers = "localhost:9092",
        AutoOffsetReset = AutoOffsetReset.Earliest,
        ClientId = "my-app",
        GroupId = "my-group",
        BrokerAddressFamily = BrokerAddressFamily.V4,
};

Here we have the Kafka URL, which is localhost:9092, as we are running Kafka in our local system. All of the configs should be similar to our producer only.

Next, we will create a consumer object and subscribe to the topic we are publishing.

using var consumer = new ConsumerBuilder<Ignore, string>(config).Build();
consumer.Subscribe("my-topic");

After that, we will start a small infinite while loop to read the message from the consumer object.

try {
    while (true) {
        var consumeResult = consumer.Consume();
        Console.WriteLine($ "Message received from {consumeResult.TopicPartitionOffset}: {
                consumeResult.Message.Value
            }
            ");
        }
    } catch (OperationCanceledException) {
        // The consumer was stopped via cancellation token.
    } finally {
        consumer.Close();
    }

So our entire method should look something like this

public void ReadMessage() {
        var config = new ConsumerConfig {
            BootstrapServers = "localhost:9092",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                ClientId = "my-app",
                GroupId = "my-group",
                BrokerAddressFamily = BrokerAddressFamily.V4,
        };
        using
        var consumer = new ConsumerBuilder < Ignore,
            string > (config).Build();
        consumer.Subscribe("my-topic");
        try {
            while (true) {
                var consumeResult = consumer.Consume();
                Console.WriteLine($ "Message received from {consumeResult.TopicPartitionOffset}: {
                        consumeResult.Message.Value
                    }
                    ");
                }
            } catch (OperationCanceledException) {
                // The consumer was stopped via cancellation token.
            } finally {
                consumer.Close();
            }
            Console.ReadLine();
        }

Now let's try to run both our applications

Consumer application first

Let's run the Producer application first

Let's have both these applications side by side

Now let's type a message in the producer window, and it should pop up in the consumer application automatically.

As you can see, we also get the same message in the consumer application window.

The above code subscribes to the "my-topic" Kafka topic and starts an infinite loop to poll for new messages. The consumer will automatically commit the offset for each consumed message. Note that the consumer will keep running until it is explicitly stopped.

You can further customize the Kafka consumer behavior by setting additional configuration options, such as the maximum number of messages to poll at once, the maximum time to wait for new messages, and the message deserialization settings.

Summary

In today's article, we saw how to implement Kafka with dotnet. If you wish to download the code, please click here for producer and here for consumer.

Happy Coding!