Kafka consumer: когда фиксировать offset и откуда берутся дубли

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

У каждой записи в партиции есть offset. Consumer хранит две разные позиции: текущую после poll() и зафиксированную для восстановления. Если обработана запись с offset 42, фиксировать нужно позицию следующей записи, обычно 43. При перезапуске consumer начнёт с сохранённой позиции. В компактированном журнале offsets могут идти с пропусками, поэтому в полноценном клиенте удобнее брать next offset из полученных records, а не всегда прибавлять единицу вручную.

Два порядка действий — два разных сбоя

Если сначала зафиксировать позицию после записи 42, а затем упасть до записи результата, при восстановлении обработчик её не увидит: платёж потерян. Поэтому offset нельзя продвигать только потому, что poll() уже вернул запись.

Если сначала сохранить результат и упасть до фиксации новой позиции, запись 42 придёт снова. Без защиты оплата будет записана дважды. Рабочий порядок для внешней БД: завершить её транзакцию, затем зафиксировать offset, а повтор сделать безопасным для результата.

Как выглядит устойчивый эффект

В PostgreSQL создадим таблицу processed_payments(payment_id TEXT PRIMARY KEY, amount_kopeks BIGINT NOT NULL). Ключ Kafka содержит устойчивый payment_id, значение — сумму в копейках. В примере используется клиент Kafka для Java, JDBC URL берётся из PAYMENTS_JDBC_URL, адрес брокера — из KAFKA_BOOTSTRAP. Consumer получает записи пачками, но завершает транзакцию и фиксирует следующую позицию отдельно для каждой партиции.

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

public class PaymentConsumer {
    public static void main(String[] args) throws SQLException {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv("KAFKA_BOOTSTRAP"));
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "payment-writer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
             Connection db = DriverManager.getConnection(System.getenv("PAYMENTS_JDBC_URL"))) {
            consumer.subscribe(Collections.singletonList("payments"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                for (TopicPartition partition : records.partitions()) {
                    db.setAutoCommit(false);
                    try {
                        for (ConsumerRecord<String, String> record : records.records(partition)) {
                            try (PreparedStatement stmt = db.prepareStatement(
                                    "INSERT INTO processed_payments(payment_id, amount_kopeks) " +
                                    "VALUES (?, ?) ON CONFLICT (payment_id) DO NOTHING")) {
                                stmt.setString(1, record.key());
                                stmt.setLong(2, Long.parseLong(record.value()));
                                stmt.executeUpdate();
                            }
                        }
                        db.commit();
                    } catch (SQLException | RuntimeException e) {
                        db.rollback();
                        throw e;
                    }
                    consumer.commitSync(Collections.singletonMap(
                            partition, records.nextOffsets().get(partition)));
                }
            }
        }
    }
}

ON CONFLICT (payment_id) DO NOTHING не создаст вторую строку при повторном чтении. Если запись в БД завершилась ошибкой, транзакция откатывается и offset не фиксируется. Если БД уже сохранила оплату, а commitSync не успел или завершился ошибкой, запись может прийти снова, но уникальный ключ сохранит один результат. Когда обработчик меняет другие таблицы, их изменения и отметка об обработанном платеже должны оставаться в одной транзакции.

С автоматической фиксацией offset это окно контролировать сложнее: commit может относиться к уже выданным через poll() записям, которые бизнес-обработчик ещё не завершил. Для такого сценария выключают auto commit и фиксируют только обработанные позиции. Если poll() выдаёт пачку или работа распараллелена, нельзя продвигать offset за незавершённую запись в той же партиции: новая позиция скрыла бы её после сбоя. Следите также за потерей владения партицией при rebalance — commit может завершиться ошибкой, и повтор возможен.

Такой порядок даёт доставку «как минимум один раз» до внешнего эффекта. Транзакции Kafka помогают атомарно связывать чтение и запись обратно в Kafka при соответствующей настройке, но сами по себе не делают произвольную запись во внешнюю базу точно однократной. Границу гарантии всегда проводите по конечному действию, которое важно бизнесу.

При разборе инцидента запишите три факта для одной партиции: offset спорной записи, сохранённый offset группы и наличие результата в целевой системе. Если результат есть, а сохранённая позиция всё ещё указывает на запись, повтор ожидаем. Если позиция уже ушла вперёд, а результата нет, ищите преждевременный commit или ошибку записи, принятую за успех. Одного сообщения «consumer получил событие» в логе недостаточно: poll() мог сдвинуть текущую позицию без сохранения её для перезапуска.

Уникальный ключ эффекта нужно выбрать по смыслу события. Offset годится как технический идентификатор записи внутри конкретной партиции, но повторно опубликованная оплата получит новый offset. Если бизнес-действие должно произойти один раз на платёж, ключом защиты должен быть устойчивый payment_id. Тогда повтор доставки и повтор публикации проходят через одно правило.

В курсе Data Engineering Fast‑Track события проходят дальше через Kafka, Spark и lakehouse под управлением Airflow. Студент собирает локальный конвейер, настраивает темы, ключи и партиции, затем проверяет свежесть данных и сценарий backfill. Откройте программу, чтобы связать поведение consumer с последующим пайплайном обработки.

Для следующего шага полезен разбор дублей при Airflow backfill: повтор там возникает уже при переигрывании интервала расчёта.