Для реализации exactly-once семантики при чтении из Kafka и сохранении данных в базу данных на уровне консьюмера в Java обычно используют транзакции и идемпотентность.

Основные подходы:

Использование Kafka Consumer с ручным коммитом оффсетов после успешной записи в БД.
- Сначала обрабатываете сообщение и сохраняете в БД.
- После успешной записи коммитите оффсет.
- Но это не гарантирует exactly-once, так как при сбое между записью и коммитом может произойти дублирование.
Идемпотентная запись в БД.
- Добавляете уникальный ключ (например, ID сообщения из Kafka) в таблицу.
- При повторной попытке записи с тем же ключом операция не изменит данные.
Использование транзакций Kafka и транзакций в БД (Two-phase commit).
- Kafka поддерживает транзакционные продюсеры, но для консьюмеров это сложнее.
- Можно использовать внешние системы, поддерживающие транзакции, или специальные библиотеки (например, Debezium, Kafka Connect с транзакционными коннекторами).
Использование схемы "consume-process-produce" с транзакционным продюсером Kafka.
- Консьюмер читает сообщение, обрабатывает и отправляет результат в другую топик транзакционно.
- Другой сервис читает из этого топика и сохраняет в БД с идемпотентностью.

Пример идемпотентной записи в БД на Java с использованием уникального ключа из Kafka:

String messageId = record.key(); // уникальный ключ сообщения
try (PreparedStatement ps = connection.prepareStatement(
    "INSERT INTO processed_messages (message_id, data) VALUES (?, ?) ON CONFLICT (message_id) DO NOTHING")) {
    ps.setString(1, messageId);
    ps.setString(2, record.value());
    ps.executeUpdate();
}
// После успешной записи коммитим оффсет
consumer.commitSync();

Таким образом, комбинация идемпотентной записи и ручного коммита оффсетов позволяет добиться exactly-once на уровне консьюмера.

Как в Kafka сделать exactly-once на уровне консьюмера при сохранении в БД (избежать дубликатов)?