Building Microservices with .NET Core and Kafka: Order Processing

In this guide, we'll walk through the process of building microservices using .NET Core and Apache Kafka, with a focus on a real-time order processing scenario. By leveraging Kafka for messaging and event-driven architecture, we'll create resilient and scalable microservices capable of handling orders in real time.

Setting Up Apache Kafka

• Install Apache Kafka on your local machine or cloud environment. Ensure that Zookeeper and Kafka brokers are running.
• Create Kafka topics for order events, such as orders, order updates, and order notifications.

Initializing .NET Core Microservices

• Create a new .NET Core solution named OrderProcessingSystem. Inside, create projects for OrderService, PaymentService, and NotificationService.
• Define the domain model for orders, payments, and notifications in each microservice.

Integrating Kafka with .NET Core

• Add the Confluent Kafka .NET client library to each microservice project via NuGet.
• Configure Kafka producer and consumer clients in each microservice to connect to Kafka brokers and topics.

Implementing Kafka Producers and Consumers

• In the OrderService, implement a Kafka producer to publish order events (e.g., OrderPlaced, OrderUpdated) to the orders topic.
• In the PaymentService and NotificationService, develop Kafka consumers to subscribe to the orders topic and process incoming order events asynchronously.

Real-Time Order Processing Workflow

• When a new order is placed, the OrderService publishes an OrderPlaced event to the orders topic.
• The PaymentService receives the OrderPlaced event, processes the payment, and publishes an OrderPaid event to the order-updates topic.
• The NotificationService subscribes to order updates and sends order confirmation notifications to customers.

Ensuring Fault Tolerance and Reliability

• Implement error handling and retry mechanisms in Kafka consumers to handle transient failures.
• Configure Kafka producers with appropriate acknowledgments to ensure message durability and reliability.

Testing and Deployment

• Write unit tests and integration tests for each microservice to validate functionality and Kafka integration.
• Deploy microservices and Kafka clusters using containerization (e.g., Docker) and orchestration tools (e.g., Kubernetes).

Here's the end-to-end code for the real-time order processing example using .NET Core and Kafka:

OrderService

using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Threading.Tasks;

public class OrderService
{
    private readonly ProducerConfig _config;
    private readonly IProducer<string, string> _producer;

    public OrderService()
    {
        _config = new ProducerConfig { BootstrapServers = "localhost:9092" };
        _producer = new ProducerBuilder<string, string>(_config).Build();
    }

    public async Task PublishOrderPlacedEvent(Order order)
    {
        var orderPlacedEvent = new OrderPlacedEvent(order);
        var message = new Message<string, string>
        {
            Key = order.Id.ToString(),
            Value = JsonConvert.SerializeObject(orderPlacedEvent)
        };

        await _producer.ProduceAsync("orders", message);
    }
}

PaymentService

using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Threading.Tasks;

public class PaymentService
{
    private readonly ConsumerConfig _config;
    private readonly IConsumer<string, string> _consumer;

    public PaymentService()
    {
        _config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "payment-group",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };
        _consumer = new ConsumerBuilder<string, string>(_config).Build();
        _consumer.Subscribe("orders");
    }

    public async Task ConsumeOrderEvents()
    {
        while (true)
        {
            try
            {
                var consumeResult = _consumer.Consume();
                var orderPlacedEvent = JsonConvert.DeserializeObject<OrderPlacedEvent>(consumeResult.Message.Value);
                // Process payment logic
                Console.WriteLine($"Payment processed for order: {orderPlacedEvent.Order.Id}");
            }
            catch (ConsumeException ex)
            {
                Console.WriteLine($"Error occurred: {ex.Error.Reason}");
            }
        }
    }
}

NotificationService

using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Threading.Tasks;

public class NotificationService
{
    private readonly ConsumerConfig _config;
    private readonly IConsumer<string, string> _consumer;

    public NotificationService()
    {
        _config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "notification-group",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };
        _consumer = new ConsumerBuilder<string, string>(_config).Build();
        _consumer.Subscribe("order-updates");
    }

    public async Task ConsumeOrderEvents()
    {
        while (true)
        {
            try
            {
                var consumeResult = _consumer.Consume();
                var orderPaidEvent = JsonConvert.DeserializeObject<OrderPaidEvent>(consumeResult.Message.Value);
                // Send notification logic
                Console.WriteLine($"Notification sent for order payment: {orderPaidEvent.Order.Id}");
            }
            catch (ConsumeException ex)
            {
                Console.WriteLine($"Error occurred: {ex.Error.Reason}");
            }
        }
    }
}

OrderPlacedEvent

public class OrderPlacedEvent
{
    public Order Order { get; set; }

    public OrderPlacedEvent(Order order)
    {
        Order = order;
    }
}

OrderPaidEvent

public class OrderPaidEvent
{
    public Order Order { get; set; }

    public OrderPaidEvent(Order order)
    {
        Order = order;
    }
}

Order

public class Order
{
    public int Id { get; set; }
    public string CustomerName { get; set; }
    public decimal Amount { get; set; }
}

This code demonstrates the implementation of a real-time order processing system using .NET Core and Kafka, including the OrderService, PaymentService, NotificationService, and relevant data models.

Conclusion

By following this real-time order processing example, you can build microservices using .NET Core and Kafka that are capable of handling complex workflows in distributed systems. Leveraging Kafka's messaging capabilities and .NET Core's flexibility, you can create resilient and scalable microservices architectures that meet the demands of modern applications.

Comments

[Mahmoud Alaskalany]

Good tutorial to show minimal kafka settings

In terms of micro services it would be very good to see in future tutorials the consumer is a standalone Console app that has a minimal http host inside and can consume the events

So it can be decoupled from the api service itself

[DotNet Full Stack Dev]

Thanks for your kind words @mahmoudalaskalany ! For now demonstrating concepts via simple application by demo the actual concept behind it. In future will sure work on more clear architecture!