10 Days of .Net Aspire: Day 5 — Integrating Apache Kafka

Step-by-step guide on how to use the .Net Aspire Kafka component in Visual Studio.

.Net Aspire framework is used to develop cloud and production-ready distributed applications. It consists of components to handle cloud-native concerns such as Redis, Postgres etc.

Prerequisites

• Install .Net 8
• Install Visual Studio 2022 version 17 or higher
• .Net Aspire Workload
• Container runtime such as Docker Desktop

Objectives

Learn how to create a starter project using .Net Aspire with the Apache Kafka component.

Github Sample: The solution structure is divided into the following projects:

• DotnetAspireChallenge.ApiService
• DotnetAspireChallenge.AppHost
• DotnetAspireChallenge.ServiceDefaults
• DotnetAspireChallenge.Web

Getting Started

Step 1: Install the following NuGet package

Install the following Nuget package into the subsequent project “DotnetAspireChallenge.AppHost”

dotnet add package Aspire.Hosting.Kafka

In the above project, register Kafka UI as shown below

    var messaging = builder.AddKafka("messaging")
```                           .WithKafkaUI();

Then finally add a reference to both the Producer and Consumer where the producer is “DotnetAspireChallenge.ApiService” and the consumer is “DotnetAspireChallenge.Web” project respectively.
```csharp
    var apiService = builder.AddProject<Projects.DotnetAspireChallenge_ApiService>("apiservice")
        .WithReference(messaging);
    
    
    
    builder.AddProject<Projects.DotnetAspireChallenge_Web>("webfrontend")
        .WithExternalHttpEndpoints()
        .WithReference(cache)
        .WithReference(apiService)
        .WithReference(messaging);

Step 2: Add dependency of Kafka Producer

Add the dependency in the Program.cs file of the project “**DotnetAspireChallenge.**ApiService”

    builder.AddKafkaProducer<string, string>("messaging");

and add a relevant minimal API endpoint using the following code.

    public static class AspireKafkaExtension
    {
        public static void MapAspireKafkaEndpoint(this WebApplication app)
        {
            app.MapGet("/send", async (IProducer<string, string> services, string key, string value) =>
            {
                try
                {
                    var message = new Message<string, string> { Key = key, Value = value };
                    DeliveryResult<string, string>? result = await services.ProduceAsync("messaging", message);
                    return result;
                }
                catch (Exception ex)
                {
    
                    throw;
                }
    
            });
        }
    }

The endpoint takes two parameters namely key and value as route values, and produces the message on the docker-hosted Kafka server.

https://localhost:7313/send?key=key&value=1

Step 3: Add dependency of Kafka Consumer

Now move “DotnetAspireChallenge.Web” project wherein register as a Kafka producer

    builder.AddKafkaConsumer<string, string>("messaging", options =>
    {
        options.Config.GroupId = "my-consumer-group";
        options.Config.AutoOffsetReset = AutoOffsetReset.Earliest;
        options.Config.EnableAutoCommit = false;
    });

Step 4: Create a Razor Page

Create a basic razor page named “KafkaConsumer.razor” to show the consumed message from the Kafka server.

    @page "/kafka"
    @attribute [StreamRendering(true)]
    @attribute [OutputCache(Duration = 5)]
    @using Confluent.Kafka
    <h3>KafkaConsumer</h3>
    
    @inject KafkaConsumeMessageClient kafaConsumeMessageClient
    <PageTitle>Kafka Consumed Message</PageTitle>
    
    <h1>Kafka</h1>
    
    <p>This component demonstrates showing data loaded from a backend API service.</p>
    
    @if (consumedMessage == null)
    {
        <p><em>Loading...</em></p>
    }
    else
    {
        <table class="table">
            <thead>
                <tr>
                    <th>Topic</th>
                    <th>Value</th>
                </tr>
            </thead>
            <tbody>
                <tr>
                    <td>@consumedMessage.Topic</td>
                    <td>@consumedMessage.Value</td>
                </tr>
    
            </tbody>
        </table>
    }
    @code {
        private ConsumeResult<string, string>? consumedMessage;
    
        protected override async Task OnInitializedAsync() => consumedMessage = kafaConsumeMessageClient.GetKafkaMessage();
    }

Step 5: Configure HttpCall to the ApiService

    public class KafkaConsumeMessageClient(HttpClient httpClient, IConsumer<string, string> _consumer)
    {
    
        public ConsumeResult<string, string>? GetKafkaMessage(CancellationToken cancellationToken = default)
        {
            ConsumeResult<string, string>? deliveryResult = null;
            _consumer.Subscribe("messaging");
            deliveryResult = _consumer.Consume(TimeSpan.FromSeconds(10));
    
            return deliveryResult;
        }
    }

Kafka Produce Demo

Kafka UI Demo

Kafka Consume Demo