Previously, I worked on a data synchronization service that required running a background task in a .NET 5 console application to consume Kafka. The requirements were automatic reconnection upon failure and graceful shutdown without message loss. After some trial and error, I've documented the process here.

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 much 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 benefit of using Host is that it handles process signals for you. When Docker sends a SIGTERM, it notifies your CancellationToken, so you don't need to write a blocking while(true) loop.

KafkaConsumerService.cs

This is the core component; here's 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 occurs, the exception is caught by the outer handler. After waiting 5 seconds, it re-enters ConsumeLoop, effectively recreating the Consumer and re-subscribing. This acts as a "guardian" mechanism, eliminating the need for a separate monitoring thread, which results in cleaner code.

What await Task.Yield() Does

If ExecuteAsync directly proceeds to consumer.Consume(), this call is blocking and will stall the host's startup process, preventing other HostedServices from starting. Adding Task.Yield() yields execution control, allowing subsequent code to run on a thread pool thread without hindering the host startup.

Why Auto-Commit Is Disabled

With EnableAutoCommit = false, offsets are manually committed after message processing. The benefit is that if an exception is thrown in HandleMessage, the offset won't be committed, ensuring the message can be consumed again upon restart.

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

consumer.Close() Placed in finally

Whether exiting normally or due to an exception, Close() will execute. It notifies the Kafka Broker that the consumer is leaving the consumer group, allowing the Broker to immediately trigger a Rebalance instead of waiting for the session.timeout.ms to expire.

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 exits → consumer.Close() is called in the finally block → the outer while loop detects IsCancellationRequested → ExecuteAsync returns → the process terminates.

This will not cause hanging or message loss. Note that Docker provides a default graceful shutdown period of 10 seconds. If your HandleMessage processing is slow, you can extend this time by adding stop_grace_period: 30s to your docker-compose.yml.

Things to Watch Out for in Production

Do not hardcode configurations in the code; place them in appsettings.json or a configuration center. The examples above are hardcoded for clarity.

If HandleMessage involves database operations, remember to use IServiceScopeFactory to create a scope and resolve scoped services. Since BackgroundService is registered as a Singleton, directly injecting scoped services like DbContext will cause issues.

If the consumption volume is high, consider batch consuming and committing messages in batches instead of committing after each message. This reduces interactions with the Broker.

This content is automatically translated to English. View Original