Какие знаешь стратегии доставки сообщений в Kafka?

Стратегии доставки сообщений в Apache Kafka

Apache Kafka предоставляет три основные гарантии доставки сообщений (delivery semantics), которые определяют, сколько раз сообщение будет обработано потребителем. Выбор правильной стратегии зависит от требований вашего приложения.

1. At-Most-Once (максимум один раз)

Сообщение может быть потеряно, но никогда не будет обработано дважды.

Характеристики:

• Producer не ждёт подтверждения (ack=0)
• Consumer не делает commit offset
• Самая быстрая, но самая ненадёжная

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "0"); // At-most-once

Producer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
producer.send(record); // Fire and forget
producer.close();

Используется когда: некритичная логика (логирование событий, метрики, аналитика)

2. At-Least-Once (минимум один раз)

Сообщение гарантированно будет доставлено, но может быть обработано несколько раз.

Характеристики:

• Producer ждёт подтверждения от брокера (ack=1 или ack=all)
• Consumer делает commit offset ПОСЛЕ обработки
• Требует идемпотентности обработки

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("acks", "1"); // At-least-once
producerProps.put("retries", 3);
producerProps.put("max.in.flight.requests.per.connection", 1);

Producer<String, String> producer = new KafkaProducer<>(producerProps);
ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
producer.send(record, new Callback() {
  public void onCompletion(RecordMetadata metadata, Exception exception) {
    if (exception != null) {
      // Retry
    } else {
      System.out.println("Message sent to partition " + metadata.partition());
    }
  }
});

// Consumer side
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "my-group");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("enable.auto.commit", "false"); // Manual commit

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("topic"));

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    processMessage(record);
  }
  consumer.commitSync(); // Commit ПОСЛЕ обработки
}

Используется когда: большинство приложений, где дублирование можно обработать (идемпотентные операции)

3. Exactly-Once (ровно один раз)

Сообщение обрабатывается ровно один раз, без потерь и дублирования.

Характеристики:

• Producer с идемпотентностью включённой
• Transactional processing
• Самая медленная, но самая надёжная

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("acks", "all"); // Все in-sync replicas должны подтвердить
producerProps.put("enable.idempotence", "true"); // Exactly-once
producerProps.put("transactional.id", "unique-id-123");

Producer<String, String> producer = new KafkaProducer<>(producerProps);
producer.initTransactions();

try {
  producer.beginTransaction();
  ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
  producer.send(record);
  producer.commitTransaction();
} catch (ProducerFenced | OutOfOrderSequenceException | AuthorizationException e) {
  producer.abortTransaction();
}

// Consumer с изолированием транзакций
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "my-group");
consumerProps.put("isolation.level", "read_committed"); // Читаем только committed messages
consumerProps.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("topic"));

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    processMessageIdempotent(record);
  }
  consumer.commitSync();
}

Сравнительная таблица

Практические рекомендации

At-Most-Once:

• Метрики и мониторинг
• Логирование некритичных событий
• Аналитика

At-Least-Once:

• Финансовые транзакции (если операция идемпотентна)
• Обработка заказов
• Уведомления пользователей

Exactly-Once:

• Критичные финансовые операции
• Сложная обработка состояния
• Где каждое сообщение имеет критичное значение

Выбор стратегии доставки — критичное решение при проектировании систем обработки потоков данных.