Previously, I was working on a data synchronization service that required running a background task in a .NET 5 console application to consume Kafka. The requirements were that it should automatically reconnect if it went down and shut down gracefully without losing messages when closed. After spending some time figuring it out, here are my notes.

Dependencies

dotnet add package Confluent.Kafka
dotnet add package Microsoft.Extensions.Hosting

Confluent.Kafka is the most widely used Kafka client in .NET, and Microsoft.Extensions.Hosting provides BackgroundService, which is far more reliable for hosting background tasks than manually creating a new Thread.

Program.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.AddHostedService<KafkaConsumerService>();
    })
    .Build()
    .Run();

One advantage of using Host is that it handles process signals for you. When Docker sends a SIGTERM, it notifies your CancellationToken, so you don't have to write a while(true) loop to block.

KafkaConsumerService.cs

This is the core part. Here is the complete code:

using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

public class KafkaConsumerService : BackgroundService
{
    private readonly ILogger<KafkaConsumerService> _logger;
    private readonly ConsumerConfig _config;
    private const string Topic = "your-topic";

    public KafkaConsumerService(ILogger<KafkaConsumerService> logger)
    {
        _logger = logger;
        _config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "your-consumer-group",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false,
            SessionTimeoutMs = 10000,
            HeartbeatIntervalMs = 3000
        };
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Task.Yield();

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ConsumeLoop(stoppingToken);
            }
            catch (OperationCanceledException)
            {
                // 正常退出，不用管
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Kafka 消费异常，5 秒后重连");
                try
                {
                    await Task.Delay(5000, stoppingToken);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
            }
        }
    }

    private async Task ConsumeLoop(CancellationToken stoppingToken)
    {
        using var consumer = new ConsumerBuilder<Ignore, string>(_config)
            .SetErrorHandler((_, e) =>
            {
                _logger.LogWarning("Kafka 错误: {Reason}", e.Reason);
            })
            .Build();

        consumer.Subscribe(Topic);
        _logger.LogInformation("开始消费 Topic: {Topic}", Topic);

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                var result = consumer.Consume(stoppingToken);
                if (result?.Message == null) continue;

                try
                {
                    await HandleMessage(result.Message.Value);
                    consumer.Commit(result);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "处理消息失败: {Value}",
                        result.Message.Value);
                }
            }
        }
        finally
        {
            consumer.Close();
        }
    }

    private Task HandleMessage(string message)
    {
        _logger.LogInformation("收到消息: {Message}", message);
        // 业务逻辑写这里
        return Task.CompletedTask;
    }
}

A Few Details

How Automatic Reconnection Works

Looking at the structure in ExecuteAsync, the outer layer is a while loop wrapping a try-catch. Inside ConsumeLoop, whether the Broker goes down, the network disconnects, or a Rebalance issue occurs, the exception is caught by the outer layer. It then waits 5 seconds and re-enters ConsumeLoop, which means recreating the Consumer and re-subscribing. This is the "guardian" pattern; it avoids the need to spawn a separate thread for monitoring, making the implementation cleaner.

What await Task.Yield() Does

If ExecuteAsync proceeds directly to consumer.Consume() upon entry, this call is blocking and would stall the host's startup process, preventing other HostedService instances from starting. Adding Task.Yield() yields execution rights, allowing the subsequent code to run on a thread pool thread, thus not interfering with the host startup.

Why Automatic Commit is Disabled

EnableAutoCommit = false means that after message processing is complete, you manually call Commit. The benefit is that if HandleMessage throws an exception, the offset is not committed, so the message can be consumed again after the next restart.

The downside is that if your business logic is not idempotent, duplicate consumption could cause issues. If your logic is idempotent or you don't mind occasionally losing one or two messages, you can simply set EnableAutoCommit to true to simplify things.

consumer.Close() in the finally Block

Whether the exit is normal or due to an exception, Close() will be executed. It notifies the Kafka Broker that this consumer is leaving the consumer group, allowing the Broker to trigger a Rebalance immediately without waiting for the session.timeout.ms timeout.

Shutdown Chain in Docker

docker stop sends a SIGTERM signal to the container. The Host listens for this signal by default → stoppingToken is canceled → consumer.Consume() throws an OperationCanceledException → the loop is exited → consumer.Close() is called in the finally block → the outer while loop checks IsCancellationRequested → ExecuteAsync returns → the process exits.

No blocking or message loss. Note that Docker provides a default 10-second graceful shutdown period. If your HandleMessage processing is relatively slow, you can add stop_grace_period: 30s to docker-compose.yml to extend it.

Things to Note in Production

Don't hardcode configuration values in your code; place them in appsettings.json or a configuration center. The hardcoded values shown above are for clarity only.

If HandleMessage involves database operations, remember to use IServiceScopeFactory to create a Scope for resolving Scoped services. BackgroundService itself is Singleton, so directly injecting Scoped services like DbContext will cause issues.

If the message consumption volume is high, consider implementing batch consumption + batch commits. You don't need to Commit for every single message, which helps reduce interactions with the Broker.