Как обрабатывать повторное чтение сообщений, если часть уже обработана, а часть нет?

Что проверяют

Понимаете ли вы, что брокеры дают доставку «хотя бы один раз», поэтому повторы неизбежны и защищаться от них должен обработчик. Следом спрашивают, что будет, если сервис упал между записью в базу и подтверждением сообщения.

Как отвечать

Сначала назовите, откуда берутся повторы. В Kafka смещение (offset) фиксируется после обработки: если консьюмер обработал 60 сообщений из пачки в 100 и упал до коммита, после перезапуска он прочитает все 100 заново. В RabbitMQ неподтверждённые сообщения вернутся в очередь с флагом Redelivered.

Каркас решения:

• идемпотентность каждого сообщения: у сообщения есть стабильный MessageId, а сама операция по возможности сформулирована как «установить», а не «прибавить»: upsert, INSERT ... ON CONFLICT DO NOTHING;
• inbox-таблица: processed_messages с уникальным ключом на MessageId. Запись в неё и бизнес-изменения идут в одной транзакции базы. Нарушение уникальности означает «уже обработано», сообщение просто подтверждается;
• подтверждение после обработки: в RabbitMQ autoAck: false и ack после коммита транзакции. В Confluent.Kafka отключают EnableAutoOffsetStore и вызывают StoreOffset только для обработанных сообщений, тогда при повторе перечитается минимум;
• многошаговая обработка: если сообщение порождает несколько побочных эффектов, храните состояние шагов или передавайте ключ идемпотентности во внешние API, чтобы повторный платёж не прошёл;
• ядовитые сообщения: ограничить число попыток и отправлять остальное в DLQ, иначе одно битое сообщение блокирует очередь или партицию.

В .NET многое из этого есть готовым: MassTransit умеет transactional outbox и inbox поверх EF Core.

Где спотыкаются

Хранят обработанные id в HashSet в памяти. После рестарта или при втором экземпляре сервиса защиты нет.

Проверяют «уже обработано?» и записывают отметку отдельными шагами вне транзакции. Падение между ними даёт повторную обработку, а два параллельных обработчика проходят проверку одновременно.

Обещают exactly-once. В Kafka он есть только для цепочки «прочитал из Kafka, записал в Kafka», а как только в деле внешняя база, нужна идемпотентность.