Kafka в .NET: Confluent.Kafka, сериализация, Schema Registry

↑ 5.4 Очереди и брокеры: Kafka, RabbitMQ · ← Предыдущая · → Следующая

Обязательно~4 мин чтенияУровень: middle

Зачем это на собесе

Как правильно написать потребителя: коммит offset, обработка ошибок, схемы сообщений.

Объяснение

Клиент Confluent.Kafka — обёртка над librdkafka.

public class OrdersConsumer(IServiceScopeFactory scopes, ILogger<OrdersConsumer> log) : BackgroundService
{
    protected override Task ExecuteAsync(CancellationToken ct) => Task.Run(async () =>
    {
        var cfg = new ConsumerConfig
        {
            BootstrapServers = "kafka:9092",
            GroupId = "billing",
            EnableAutoCommit = false,               // коммитим вручную после обработки
            AutoOffsetReset = AutoOffsetReset.Earliest,
        };
        using var consumer = new ConsumerBuilder<string, string>(cfg).Build();
        consumer.Subscribe("orders");
        try
        {
            while (!ct.IsCancellationRequested)
            {
                var cr = consumer.Consume(ct);
                using var scope = scopes.CreateScope();
                await scope.ServiceProvider.GetRequiredService<IOrderHandler>().HandleAsync(cr.Message.Value, ct);
                consumer.Commit(cr);                // at-least-once
            }
        }
        catch (OperationCanceledException) { }
        finally { consumer.Close(); }               // корректный выход из группы
    }, ct);
}

Сериализация и Schema Registry

Формат	Плюсы
JSON	просто, читаемо, нет контроля схемы
Avro / Protobuf + Schema Registry	компактно, схема хранится в реестре, проверка совместимости

Schema Registry задаёт режимы совместимости (BACKWARD, FORWARD, FULL): изменение схемы, нарушающее правила, будет отклонено при регистрации.

Нюансы и подводные камни

• Consume блокирует поток: запускайте в отдельной задаче/потоке.
• Автокоммит до окончания обработки приводит к потере сообщения при падении.
• Долгая обработка → выход из группы по max.poll.interval.ms и rebalance.
• Всегда Close() при остановке, иначе группа ждёт таймаут сессии.
• Не обрабатывайте параллельно сообщения одной партиции без учёта порядка.

Практика

1. Напишите потребителя с ручным коммитом и корректной остановкой.
2. Добавьте Schema Registry и Avro/Protobuf.
3. Сымитируйте падение обработчика и проверьте повторную обработку.

Вопросы с ответами

Почему отключают автокоммит?

Чтобы коммитить offset только после успешной обработки и не терять сообщения.

Зачем Schema Registry?

Централизованное хранение схем и проверка совместимости при их эволюции.

Что будет при долгой обработке?

Потребитель может быть исключён из группы (max.poll.interval.ms), начнётся rebalance и сообщение обработается повторно.

Связанные темы

• Kafka: producer, acks, идемпотентность, порядок сообщений
• RabbitMQ: exchanges, queues, routing, ack