Event-Driven Order System with Kafka

Build two .NET microservices that communicate through Apache Kafka.

Introduction

30 Second Summary

When you order something online, you don't wait on the checkout page while the warehouse physically picks your item off the shelf. The store confirms your order instantly and the fulfillment happens separately, behind the scenes.

In this project, you will build two microservices that communicate through Apache Kafka, replicating this same pattern in code. Your Order API will accept orders instantly while a separate Inventory Service processes them at its own pace.

What You'll Build

An event-driven order processing system where an ASP.NET Core API publishes order events to Kafka and a .NET Worker Service consumes those events to update stock levels in SQL Server.

By the end of this project, you'll have:

  • A Docker Compose environment running Kafka in KRaft mode and SQL Server locally.
  • An ASP.NET Core Minimal API that publishes order events to a Kafka topic.
  • A .NET Worker Service that consumes Kafka events and updates product stock in a database using Entity Framework Core.
  • Secret Mission: Add a dead-letter topic that captures failed messages when inventory is insufficient.

Are there any prerequisites?

Basic programming knowledge is all you need. Familiarity with REST APIs is helpful but not required.

Set Up Docker Infrastructure

Before you can build microservices that talk to each other through events, you need infrastructure for them to run on.

In this step, you will spin up a Kafka broker and a SQL Server database as Docker containers, then scaffold the .NET solution that your two services will live in.

In this step, get ready to:
  • Create a Docker Compose file that runs Kafka and SQL Server locally.
  • Verify both containers are healthy and accepting connections.
  • Scaffold a .NET solution with three projects and install NuGet packages.
Create the Docker Compose file

Your two microservices need a message broker to pass events between them and a database to store inventory. Docker Compose lets you define both services in a single file and start them with one command.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir order-system
cd order-system
  • Confirm the folder exists by running:
ls

You should see the order-system folder listed in the output (it will be empty for now).

  • Create the docker-compose.yml file inside your order-system folder by running:
New-Item docker-compose.yml
  • Open the order-system folder in VS Code by pressing the Windows key, typing VS Code, and pressing Enter.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop > order-system, then click Select Folder.
  • In the Explorer sidebar, click docker-compose.yml to open it.
  • Paste the following configuration into docker-compose.yml:
services:
  broker:
    image: apache/kafka:latest
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  sqlserver:
    image: mcr.microsoft.com/mssql/server:2022-latest
    container_name: sqlserver
    ports:
      - "1433:1433"
    environment:
      ACCEPT_EULA: "Y"
      MSSQL_SA_PASSWORD: "YourStrong!Passw0rd"

What does this configuration do?

This file defines two containers that Docker will manage together:

  • broker runs Apache Kafka in KRaft mode. KRaft means Kafka manages its own cluster metadata internally, so you do not need a separate ZooKeeper container. Port 9092 is exposed for your .NET services to connect from the host machine.
  • sqlserver runs SQL Server 2022 on port 1433. The MSSQL_SA_PASSWORD environment variable sets the admin password.

Why KRaft mode instead of ZooKeeper?

Older Kafka setups required a separate ZooKeeper container to manage cluster metadata. Kafka 4.x removed the ZooKeeper dependency entirely. KRaft mode is simpler (one container instead of two), starts faster, and is the current recommended approach for local development.

  • Save docker-compose.yml by pressing Ctrl+S.
  • Start both containers by running this command in your terminal:
docker compose up -d

What does this command do?

The up subcommand creates and starts the containers defined in your Compose file. The -d flag runs them in the background so your terminal stays free.

  • Verify both containers are running by executing:
docker ps

You should see two containers listed with a status of Up: one named broker and one named sqlserver.

Containers not showing as Up?

  • Make sure Docker Desktop is running. You can check by looking for the Docker whale icon in your system tray.
  • If the sqlserver container shows Exited, the password may not meet complexity requirements. The password YourStrong!Passw0rd in the Compose file satisfies SQL Server's policy (uppercase, lowercase, digit, and symbol).
  • If the broker container exited, run docker logs broker to check for port conflicts. Another service may already be using port 9092.

Still stuck? Help me troubleshoot my Docker containers not starting.

Verify Kafka and SQL Server are accessible

Running containers do not always mean working services. Kafka needs a few seconds to elect a controller, and SQL Server needs time to initialize its system databases. You will create a test topic in Kafka and query SQL Server's version to confirm both are ready.

  • Wait about 10 seconds for the services to finish starting, then create a test topic in Kafka by running:
docker exec broker /opt/kafka/bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092

You should see a message confirming the topic was created.

What does this command do?

This runs the kafka-topics.sh script inside the running broker container. The --create flag tells Kafka to register a new topic called test-topic. If Kafka is healthy, it responds with a success message.

  • Verify SQL Server is accepting connections by running:
docker exec sqlserver /opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P "YourStrong!Passw0rd" -No -Q "SELECT @@VERSION"

You should see output containing the SQL Server 2022 version string.

What does this command do?

This executes sqlcmd inside the sqlserver container. The -No flag makes encryption optional (required for newer sqlcmd versions connecting to a container with a self-signed certificate). SELECT @@VERSION returns the installed SQL Server version, proving the database is up and accepting queries.

Seeing a connection error?

  • SQL Server can take 15-20 seconds to fully start. Wait a moment and retry the command.
  • If the Kafka command fails, the broker may still be initializing. Run docker logs broker and look for a line mentioning the controller being elected.

Need more help? Help me troubleshoot connections to my Kafka and SQL Server containers.

Create the .NET solution structure

Your project needs three separate .NET projects: an API that receives orders, a worker that processes them, and a shared class library for the event contract both services use.

  • Create the .NET solution and all three projects by running these commands:
dotnet new sln -n OrderSystem
dotnet new webapi -n OrderApi
dotnet new worker -n InventoryService
dotnet new classlib -n Events

What are these projects?

  • OrderApi is an ASP.NET Core Minimal API that will accept HTTP requests and publish events to Kafka.
  • InventoryService is a .NET Worker Service that will run a background loop, consuming events from Kafka and updating the database.
  • Events is a class library that defines the shared event contract. Both services will reference it so they agree on the shape of messages.
  • Add all three projects to the solution by running:
dotnet sln add OrderApi/OrderApi.csproj
dotnet sln add InventoryService/InventoryService.csproj
dotnet sln add Events/Events.csproj
  • Add the Confluent.Kafka NuGet package to both the OrderApi and InventoryService by running:
dotnet add OrderApi/OrderApi.csproj package Confluent.Kafka --version 2.14.0
dotnet add InventoryService/InventoryService.csproj package Confluent.Kafka --version 2.14.0
  • Add Entity Framework Core for SQL Server to the InventoryService by running:
dotnet add InventoryService/InventoryService.csproj package Microsoft.EntityFrameworkCore.SqlServer --version 8.0.25
  • Add project references so both services can use the shared Events library by running:
dotnet add OrderApi/OrderApi.csproj reference Events/Events.csproj
dotnet add InventoryService/InventoryService.csproj reference Events/Events.csproj

Why a shared Events library?

When two services communicate through events, both need to agree on the shape of the message. Defining the event class in a shared library means any change to the contract is visible to both producer and consumer at compile time. This prevents subtle serialization bugs.

  • Verify the solution structure by running:
ls

You should see folders for OrderApi, InventoryService, Events, plus docker-compose.yml and OrderSystem.sln.

Command not found when running dotnet?

  • Make sure the .NET 8 SDK is installed. Run dotnet --version and confirm you see a version starting with 8..
  • If the version command works but package installs fail, check your internet connection. NuGet needs to download packages from nuget.org.

Still stuck? Help me fix dotnet CLI errors during project setup.

Now delete the boilerplate files that the templates generated so you have a clean slate. The dotnet new webapi template creates sample files you do not need.

  • Remove the template-generated files by running:
Remove-Item OrderApi/OrderApi.http
Remove-Item InventoryService/Worker.cs
Remove-Item Events/Class1.cs

✔️ Awesome, I've got everything!

Great. Double check you have saved docker-compose.yml and that all three project folders exist.

ⓧ I'd like to double check the full code

Here is the full docker-compose.yml and the three .csproj files as they should look after all packages and references are added:

services:
  broker:
    image: apache/kafka:latest
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  sqlserver:
    image: mcr.microsoft.com/mssql/server:2022-latest
    container_name: sqlserver
    ports:
      - "1433:1433"
    environment:
      ACCEPT_EULA: "Y"
      MSSQL_SA_PASSWORD: "YourStrong!Passw0rd"
<Project Sdk="Microsoft.NET.Sdk.Web">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Confluent.Kafka" Version="2.14.0" />
  </ItemGroup>

  <ItemGroup>
    <ProjectReference Include="..\Events\Events.csproj" />
  </ItemGroup>

</Project>
<Project Sdk="Microsoft.NET.Sdk.Worker">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Confluent.Kafka" Version="2.14.0" />
    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="8.0.25" />
  </ItemGroup>

  <ItemGroup>
    <ProjectReference Include="..\Events\Events.csproj" />
  </ItemGroup>

</Project>
<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>

</Project>

Your local infrastructure is live and your .NET solution is scaffolded. Next up, you will build the Order API that publishes events to Kafka when a new order comes in.

Build the Order API with Kafka Producer

Your Docker infrastructure is ready with Kafka and SQL Server running in containers. Now it's time to build the first microservice that actually produces events.

The Order API accepts incoming HTTP requests and publishes an event to Kafka every time someone places an order. Instead of calling the Inventory Service directly, the Order API simply announces "an order was placed" and moves on. This is the core of event-driven architecture.

In this step, get ready to:
  • Define the shared OrderPlaced event contract.
  • Build a Minimal API with a Kafka producer that publishes order events.
  • Test the full flow by sending a POST request and verifying the message in Kafka.
Define the OrderPlaced event

Both the Order API and the Inventory Service need to agree on what an "order event" looks like. The shared Events class library is the single place where this contract lives.

  • In VS Code, create a new file called OrderPlaced.cs inside the Events folder.
  • Add the following code to define the event class:
namespace Events;

public class OrderPlaced
{
    public Guid OrderId { get; set; }
    public string ProductId { get; set; } = string.Empty;
    public int Quantity { get; set; }
    public string CustomerName { get; set; } = string.Empty;
    public DateTime CreatedAt { get; set; }
}

What does this code do?

This class defines the shape of every order event that flows through Kafka. Each property represents a piece of information the downstream services need to process the order.

  • OrderId uniquely identifies each order (generated as a GUID).
  • ProductId tells the consumer which product was ordered.
  • Quantity is how many units were requested.
  • CreatedAt records when the order was placed, which is useful for auditing.
  • Save Events/OrderPlaced.cs.

Seeing a red squiggly under the namespace?

Make sure the file is saved inside the Events folder (not OrderApi or InventoryService). The namespace Events must match the project name.

help me fix namespace errors in my Events class library

Create the Order API with Kafka producer

The Order API uses ASP.NET Core Minimal APIs to define HTTP endpoints with minimal boilerplate. You will register a Kafka producer as a singleton in the dependency injection container so it can be reused across requests.

  • Open OrderApi/Program.cs in VS Code.
  • Replace the entire contents of the file with the following code:
using System.Text.Json;
using Confluent.Kafka;
using Events;

var builder = WebApplication.CreateBuilder(args);

// Register a Kafka producer as a singleton so it's reused across requests
builder.Services.AddSingleton<IProducer<string, string>>(sp =>
{
    var config = new ProducerConfig
    {
        BootstrapServers = "localhost:9092"
    };
    return new ProducerBuilder<string, string>(config).Build();
});

var app = builder.Build();

What does this code do?

This first section sets up the application and configures the Kafka producer.

  • ProducerConfig tells the producer where to find the Kafka broker (your Docker container on port 9092).
  • AddSingleton means a single producer instance is created and shared across all incoming HTTP requests. Kafka producers are thread-safe and designed to be long-lived.
  • IProducer<string, string> is the Confluent.Kafka interface for producing messages with string keys and string values.

Now add the POST endpoint and the remaining application code below the line you just wrote. Add this directly after var app = builder.Build(); in the same file:

app.MapPost("/orders", async (OrderRequest request, IProducer<string, string> producer) =>
{
    // Create the event from the incoming request
    var orderEvent = new OrderPlaced
    {
        OrderId = Guid.NewGuid(),
        ProductId = request.ProductId,
        Quantity = request.Quantity,
        CustomerName = request.CustomerName,
        CreatedAt = DateTime.UtcNow
    };

    // Serialize the event and produce it to the order-events topic
    var message = new Message<string, string>
    {
        Key = orderEvent.ProductId,
        Value = JsonSerializer.Serialize(orderEvent)
    };

    await producer.ProduceAsync("order-events", message);

    return Results.Accepted($"/orders/{orderEvent.OrderId}", orderEvent);
});

app.MapGet("/", () => "Order API is running");

app.Run();

public record OrderRequest(string ProductId, int Quantity, string CustomerName);

What does this code do?

This section defines the HTTP endpoints and the data shape for incoming requests.

  • MapPost("/orders", ...) creates a POST endpoint that accepts order details. ASP.NET Core automatically deserializes the JSON body into an OrderRequest and injects the Kafka producer from DI.
  • The Key is set to ProductId so all orders for the same product go to the same Kafka partition. This guarantees ordering per product.
  • ProduceAsync sends the serialized JSON to the order-events topic and waits for broker acknowledgment.
  • Results.Accepted returns HTTP 202, signaling that the order was accepted for processing (not completed yet).
  • The OrderRequest record at the bottom defines what the POST body must contain.

Why return 202 instead of 200?

HTTP 202 Accepted means "I received your request and will process it later." This is the standard response in event-driven systems because the order is not fully processed yet. The Inventory Service will handle it asynchronously after consuming the event from Kafka.

  • Save OrderApi/Program.cs.

Seeing errors for Confluent.Kafka or Events?

Make sure the OrderApi.csproj file includes both the Confluent.Kafka package reference and the Events project reference. These were added in the previous step.

help me fix missing references in my OrderApi project

✔️ Awesome, I've got everything!

Great. Double check you have saved OrderApi/Program.cs with all the code above.

ⓧ I'd like to double check the full code

Here is the complete OrderApi/Program.cs file:

using System.Text.Json;
using Confluent.Kafka;
using Events;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSingleton<IProducer<string, string>>(sp =>
{
    var config = new ProducerConfig
    {
        BootstrapServers = "localhost:9092"
    };
    return new ProducerBuilder<string, string>(config).Build();
});

var app = builder.Build();

app.MapPost("/orders", async (OrderRequest request, IProducer<string, string> producer) =>
{
    var orderEvent = new OrderPlaced
    {
        OrderId = Guid.NewGuid(),
        ProductId = request.ProductId,
        Quantity = request.Quantity,
        CustomerName = request.CustomerName,
        CreatedAt = DateTime.UtcNow
    };

    var message = new Message<string, string>
    {
        Key = orderEvent.ProductId,
        Value = JsonSerializer.Serialize(orderEvent)
    };

    await producer.ProduceAsync("order-events", message);

    return Results.Accepted($"/orders/{orderEvent.OrderId}", orderEvent);
});

app.MapGet("/", () => "Order API is running");

app.Run();

public record OrderRequest(string ProductId, int Quantity, string CustomerName);
Test the Order API with the Kafka console consumer

Now you will run the Order API, send a test order, and confirm the event arrives in Kafka. This proves the full producer flow is working end to end.

  • Open a terminal in VS Code by pressing Ctrl+`.
  • Start the Kafka console consumer in your terminal by running this command:
docker exec -it broker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-events --from-beginning

What does this command do?

This starts a live listener inside your Kafka broker container. It subscribes to the order-events topic and prints every message that arrives. The --from-beginning flag means it reads all messages from the start, not just new ones.

The terminal will appear to hang with no output. This is normal. It is waiting for messages to arrive.

  • Open a second terminal in VS Code by clicking the + icon in the terminal panel.
  • Navigate to the OrderApi project and start the API by running these commands:
cd OrderApi
dotnet run --urls "http://localhost:5000"

Why specify --urls?

The --urls flag pins the API to a known port (5000) so your test commands work reliably. Without it, the port depends on your generated launchSettings.json file and may vary.

Wait until you see output indicating the API is listening (a line mentioning http://localhost:5000).

  • Open a third terminal in VS Code by clicking the + icon again.
  • Send a test order to the API by running this command:
curl -X POST http://localhost:5000/orders -H "Content-Type: application/json" -d "{\"productId\":\"WIDGET-001\",\"quantity\":5,\"customerName\":\"Alice\"}"
  • Check the response in your terminal. You should see a JSON response with a 202 status containing the OrderId, ProductId, Quantity, CustomerName, and CreatedAt fields.
  • Switch to your first terminal (the one running the Kafka console consumer).
  • You should see the serialized JSON event printed in the console consumer output, confirming the message arrived in the order-events topic.

Not seeing the message in the consumer?

Make sure your Docker containers are still running. Run docker ps to confirm the broker container is up.

If the curl command returns a connection error, verify the API is running and listening on port 5000 in your second terminal.

help me troubleshoot why my Kafka producer message is not appearing in the console consumer

Your Order API is producing events to Kafka. Next up, you will build the Inventory Service that consumes these events and updates stock levels in SQL Server.

Build the Inventory Service Consumer

Your Order API is live and publishing events to Kafka every time someone places an order.

Now you need something listening on the other end. In this step, you will build the Inventory Service: a background worker that subscribes to the order-events topic, reads each order, and decrements stock levels in SQL Server. This is the consumer half of the event-driven pattern.

In this step, get ready to:
  • Create the Product entity and database context with seed data.
  • Build a background worker that consumes Kafka events and updates stock.
  • Run both services and verify end-to-end event processing.
Create the Product entity and database context

Before the Inventory Service can track stock, it needs a data model and a way to talk to SQL Server. You will create a Product entity class and an Entity Framework Core database context that seeds the Products table with sample inventory.

  • In VS Code, create a new file called Product.cs inside the InventoryService folder.
  • Add the following code to define the Product entity:
namespace InventoryService;

// Represents a product row in the Products table
public class Product
{
    public int Id { get; set; }
    public string ProductId { get; set; } = string.Empty;
    public string Name { get; set; } = string.Empty;
    public int Stock { get; set; }
}

What does this code do?

This class maps to a row in the SQL Server Products table. Each product has a unique Id (primary key), a ProductId string for business identification (like WIDGET-001), a display Name, and a Stock count that the consumer will decrement.

  • Save Product.cs.
  • Create another new file called AppDbContext.cs inside the InventoryService folder.
  • Add the following code to define the database context with seed data:
using Microsoft.EntityFrameworkCore;

namespace InventoryService;

// EF Core context that maps to the InventoryDb database
public class AppDbContext : DbContext
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

    // Table of products with stock levels
    public DbSet<Product> Products { get; set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        // Seed the database with sample inventory
        modelBuilder.Entity<Product>().HasData(
            new Product { Id = 1, ProductId = "WIDGET-001", Name = "Blue Widget", Stock = 100 },
            new Product { Id = 2, ProductId = "GADGET-002", Name = "Red Gadget", Stock = 50 },
            new Product { Id = 3, ProductId = "GIZMO-003", Name = "Green Gizmo", Stock = 25 }
        );
    }
}

What does this code do?

AppDbContext is your Entity Framework Core gateway to SQL Server. The Products DbSet exposes the table for queries and updates.

The OnModelCreating override uses HasData to seed three products into the table when the database is first created. This gives the consumer something to decrement when order events arrive.

  • Save AppDbContext.cs.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files.

ⓧ I'd like to double check the full code

Here is Product.cs:

namespace InventoryService;

public class Product
{
    public int Id { get; set; }
    public string ProductId { get; set; } = string.Empty;
    public string Name { get; set; } = string.Empty;
    public int Stock { get; set; }
}

Here is AppDbContext.cs:

using Microsoft.EntityFrameworkCore;

namespace InventoryService;

public class AppDbContext : DbContext
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

    public DbSet<Product> Products { get; set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<Product>().HasData(
            new Product { Id = 1, ProductId = "WIDGET-001", Name = "Blue Widget", Stock = 100 },
            new Product { Id = 2, ProductId = "GADGET-002", Name = "Red Gadget", Stock = 50 },
            new Product { Id = 3, ProductId = "GIZMO-003", Name = "Green Gizmo", Stock = 25 }
        );
    }
}
Implement the Kafka consumer as a background worker

A Kafka consumer is a long-running loop that polls for new messages. In .NET, the natural home for this kind of work is a BackgroundService. Your worker will subscribe to the order-events topic, deserialize each event, look up the product, and decrement stock.

  • In the InventoryService folder, open the existing Worker.cs file (the template created it for you).
  • Replace its entire contents with the following code:
using System.Text.Json;
using Confluent.Kafka;
using Events;

namespace InventoryService;

public class Worker : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<Worker> _logger;

    public Worker(IServiceProvider serviceProvider, ILogger<Worker> logger)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Configure the Kafka consumer
        var config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "inventory-service",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false
        };

        using var consumer = new ConsumerBuilder<string, string>(config).Build();
        consumer.Subscribe("order-events");

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // Poll for the next message
                var result = consumer.Consume(stoppingToken);
                var orderEvent = JsonSerializer.Deserialize<OrderPlaced>(result.Message.Value);

                if (orderEvent is null)
                {
                    _logger.LogWarning("Received null event, skipping.");
                    consumer.Commit(result);
                    continue;
                }

                _logger.LogInformation("Processing order {OrderId} for product {ProductId}, quantity {Quantity}",
                    orderEvent.OrderId, orderEvent.ProductId, orderEvent.Quantity);

                // Create a new scope to resolve the scoped DbContext
                using var scope = _serviceProvider.CreateScope();
                var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

                var product = await db.Products.FirstOrDefaultAsync(
                    p => p.ProductId == orderEvent.ProductId, stoppingToken);

                if (product is null)
                {
                    _logger.LogWarning("Product {ProductId} not found.", orderEvent.ProductId);
                    consumer.Commit(result);
                    continue;
                }

                // Decrement stock and save
                product.Stock -= orderEvent.Quantity;
                await db.SaveChangesAsync(stoppingToken);

                _logger.LogInformation("Updated stock for {ProductId}. New stock: {Stock}",
                    product.ProductId, product.Stock);

                // Manually commit the offset after successful processing
                consumer.Commit(result);
            }
            catch (ConsumeException ex)
            {
                _logger.LogError(ex, "Kafka consume error.");
            }
        }

        consumer.Close();
    }
}

What does this code do?

The ConsumerConfig sets up the connection to Kafka. GroupId identifies this consumer group so Kafka tracks which messages have been delivered. AutoOffsetReset = Earliest means if there is no prior commit, start from the beginning of the topic.

EnableAutoCommit = false is critical. It means the consumer must explicitly call consumer.Commit(result) after processing. If the service crashes before committing, Kafka will redeliver the message on the next startup. This guarantees at-least-once delivery.

The worker uses IServiceProvider.CreateScope() to get a fresh AppDbContext for each message. This is the documented pattern for resolving scoped services inside long-running hosted services.

  • Save Worker.cs.

Seeing a red underline on FirstOrDefaultAsync?

If your editor shows an error on FirstOrDefaultAsync, add using Microsoft.EntityFrameworkCore; at the top of the file. This extension method lives in the EF Core namespace.

help me fix EF Core namespace errors in my Worker.cs

Now wire up the Program.cs to configure the database connection and register the worker.

  • Open Program.cs in the InventoryService folder.
  • Replace its entire contents with the following code:
using InventoryService;
using Microsoft.EntityFrameworkCore;

var builder = Host.CreateApplicationBuilder(args);

// Register EF Core with the SQL Server connection string
builder.Services.AddDbContext<AppDbContext>(options =>
    options.UseSqlServer("Server=localhost,1433;Database=InventoryDb;User Id=sa;Password=YourStrong!Passw0rd;TrustServerCertificate=True;"));

// Register the Kafka consumer as a hosted background service
builder.Services.AddHostedService<Worker>();

var host = builder.Build();

// Ensure the database and tables exist on startup
using (var scope = host.Services.CreateScope())
{
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
    db.Database.EnsureCreated();
}

host.Run();

What does this code do?

AddDbContext registers AppDbContext with the dependency injection container, pointing it at the SQL Server running in Docker on port 1433. The connection string uses TrustServerCertificate=True because the Docker container uses a self-signed certificate.

db.Database.EnsureCreated() checks if the InventoryDb database exists and creates it (plus the Products table and seed data) if it does not. This avoids the complexity of running EF Core migrations for a learning project.

  • Save Program.cs.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all four files in the InventoryService folder.

ⓧ I'd like to double check the full code

Here is Worker.cs:

using System.Text.Json;
using Confluent.Kafka;
using Events;

namespace InventoryService;

public class Worker : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<Worker> _logger;

    public Worker(IServiceProvider serviceProvider, ILogger<Worker> logger)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "inventory-service",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false
        };

        using var consumer = new ConsumerBuilder<string, string>(config).Build();
        consumer.Subscribe("order-events");

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var result = consumer.Consume(stoppingToken);
                var orderEvent = JsonSerializer.Deserialize<OrderPlaced>(result.Message.Value);

                if (orderEvent is null)
                {
                    _logger.LogWarning("Received null event, skipping.");
                    consumer.Commit(result);
                    continue;
                }

                _logger.LogInformation("Processing order {OrderId} for product {ProductId}, quantity {Quantity}",
                    orderEvent.OrderId, orderEvent.ProductId, orderEvent.Quantity);

                using var scope = _serviceProvider.CreateScope();
                var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

                var product = await db.Products.FirstOrDefaultAsync(
                    p => p.ProductId == orderEvent.ProductId, stoppingToken);

                if (product is null)
                {
                    _logger.LogWarning("Product {ProductId} not found.", orderEvent.ProductId);
                    consumer.Commit(result);
                    continue;
                }

                product.Stock -= orderEvent.Quantity;
                await db.SaveChangesAsync(stoppingToken);

                _logger.LogInformation("Updated stock for {ProductId}. New stock: {Stock}",
                    product.ProductId, product.Stock);

                consumer.Commit(result);
            }
            catch (ConsumeException ex)
            {
                _logger.LogError(ex, "Kafka consume error.");
            }
        }

        consumer.Close();
    }
}

Here is Program.cs:

using InventoryService;
using Microsoft.EntityFrameworkCore;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddDbContext<AppDbContext>(options =>
    options.UseSqlServer("Server=localhost,1433;Database=InventoryDb;User Id=sa;Password=YourStrong!Passw0rd;TrustServerCertificate=True;"));

builder.Services.AddHostedService<Worker>();

var host = builder.Build();

using (var scope = host.Services.CreateScope())
{
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
    db.Database.EnsureCreated();
}

host.Run();
Run both services and test end-to-end

Time to see the full event-driven loop in action. You will run both the Order API and the Inventory Service simultaneously, then send an order and watch the stock decrease.

  • Open a terminal in VS Code (Ctrl+`).
  • Make sure you are in the solution root folder (the folder containing OrderApi, InventoryService, and Events subfolders).
  • Start the Inventory Service by running:
dotnet run --project InventoryService/InventoryService.csproj

You should see log output indicating the worker has started and is waiting for messages.

  • Open a second terminal in VS Code by clicking the + icon in the terminal panel.
  • Start the Order API in the second terminal by running:
dotnet run --project OrderApi/OrderApi.csproj

You should see output indicating the API is listening on a local port.

  • Open a third terminal and send a test order for 5 units of WIDGET-001 by running:
curl -X POST http://localhost:5000/orders -H "Content-Type: application/json" -d "{\"ProductId\": \"WIDGET-001\", \"Quantity\": 5, \"CustomerName\": \"Alice\"}"

Using PowerShell instead of curl?

If curl does not work in your terminal, use Invoke-RestMethod or the built-in HTTP test file in VS Code. The important thing is to send a POST to /orders with the JSON body above.

  • Switch to the first terminal (the one running the Inventory Service).

You should see log messages showing the event was consumed and stock was updated:

info: InventoryService.Worker[0]
      Processing order <guid> for product WIDGET-001, quantity 5
info: InventoryService.Worker[0]
      Updated stock for WIDGET-001. New stock: 95
  • Confirm the stock change in the database by running this command in your third terminal:
docker exec sqlserver /opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P "YourStrong!Passw0rd" -d InventoryDb -Q "SELECT ProductId, Name, Stock FROM Products" -No

You should see WIDGET-001 with a stock value of 95 (down from the original 100).

Why manual offset commits?

With EnableAutoCommit = false, the consumer only marks a message as processed after your code finishes updating the database. If the service crashes mid-processing, Kafka redelivers the message on restart. This is called at-least-once delivery and is the safest default for systems that modify state.

Not seeing messages consumed?

Check that your Docker containers are still running with docker ps. Both broker and sqlserver must be in a healthy state.

If the Inventory Service starts but never processes messages, verify the BootstrapServers in Worker.cs is localhost:9092 and that the Order API is publishing to the topic named order-events.

help me debug why my Inventory Service is not consuming Kafka messages

Your event-driven system is working end-to-end. The Order API publishes events without waiting for the Inventory Service, and the Inventory Service processes them at its own pace. Next up, you will extend this with a dead-letter queue for handling failed orders.

Secret mission

Add a Dead-Letter Topic for Failed Orders

Right now, your Inventory Service silently drops orders for products that don't exist and blindly decrements stock even when there isn't enough. In production, losing messages like this means lost revenue and confused customers. Your mission: route failed events to a dead-letter queue so nothing disappears into the void.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container: broker (Apache Kafka, image apache/kafka:latest).
  • Docker container: sqlserver (SQL Server 2022, image mcr.microsoft.com/mssql/server:2022-latest).
  • Local project folder on your Desktop containing docker-compose.yml, OrderApi, InventoryService, and Events projects.

Keep everything running

No action needed. Choose this if you're still actively building or plan to continue with the next project in this series.

  • Your broker and sqlserver containers will continue running in Docker Desktop.
  • Your project files remain on your Desktop, ready for further development.

Pause - I'll come back to this later

Stop the running containers to free up memory without removing any data. Your database records and Kafka topics are preserved.

  • Open your terminal and navigate to the project folder containing docker-compose.yml.
  • Stop both containers by running:
docker compose stop
  • When you're ready to resume, start them again by running:
docker compose start

Your containers will pick up right where they left off with all data intact.

Delete - I don't want to use this again

Remove all Docker containers, networks, and data volumes. Then delete your local project files.

  • Open your terminal and navigate to the project folder containing docker-compose.yml.
  • Remove both containers and their data volumes by running:
docker compose down -v

This removes the broker and sqlserver containers along with any stored data (your SQL Server database and Kafka logs).

  • Delete your project folder from the Desktop. On Windows, run:
cd ~\Desktop
rmdir /s /q event-driven-orders
  • On macOS or Linux, run:
cd ~/Desktop
rm -rf event-driven-orders

Nice Work!

Nice Work!

Nice work! You just built a fully functional event-driven microservices system with two .NET services communicating through Apache Kafka.

You've learned how to:

  • Built an ASP.NET Core Minimal API that serializes events and publishes them to an Apache Kafka topic.
  • Created a .NET Worker Service that consumes Kafka events, processes them, and persists inventory changes to SQL Server using Entity Framework Core.
  • Set up a fully containerized local development environment with Docker Compose, running Kafka in KRaft mode and SQL Server with zero cloud costs.
  • Secret Mission: Added a dead-letter queue that captures failed order events when inventory is insufficient, a critical production error-handling pattern.

Ready to quiz yourself?