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

Типы гарантии доставки сообщений

Гарантия доставки (Delivery Guarantee) — это один из ключевых аспектов систем обмена сообщениями. Существует три основных типа гарантий, каждый из которых обеспечивает разные уровни надёжности и имеет свои компромиссы с точки зрения производительности и сложности.

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

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

Как это работает

// Kafka: fire-and-forget
ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic", "key", "message");
kafkaProducer.send(record);  // Отправил и забыл

Сценарий:

Производитель → (отправка) → Брокер
                    ↓
                 сбой?
                 Сообщение потеряно,
                 производитель не знает

Производитель не пересылает сообщение

Реализация

// Kafka с acks=0 (At-Most-Once)
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
props.put("retries", "0");  // Не пересылать
props.put("linger.ms", "10");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = 
        new ProducerRecord<>("my-topic", "msg-" + i);
    
    // Fire and forget
    producer.send(record);  // Не проверяем результат
}

producer.close();

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

Преимущества:

• Самая высокая производительность — не ждёт подтверждений
• Минимальная задержка (latency)
• Простая реализация

Проблемы:

• Сообщения могут быть потеряны при сбое брокера
• Нет повторных попыток
• Невозможно гарантировать доставку

Когда использовать

// Подходит для:
// - Логирование (потеря одного лога — не критично)
// - Метрики (одна потеря метрики — не критично)
// - Аналитика (потеря нескольких events приемлемо)

public class AnalyticsLogger {
    public void logUserAction(String userId, String action) {
        kafkaProducer.send(
            new ProducerRecord<>(
                "user-actions",
                userId,
                action  // Если потеряется, это не критично
            )
        );
    }
}

Не подходит для:

// НЕ подходит для:
// - Платёжные операции (потеря = потеря денег)
// - Критичные бизнес-события
// - Важные уведомления пользователям

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

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

Как это работает

Производитель → отправка → Брокер
    ↑                           ↓
    └── (если нет подтверждения) ──┘
         пересылаю ещё раз

Реализация

// Kafka с acks=all (At-Least-Once)
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", "all");  // At-Least-Once
props.put("retries", "3");  // Пересылать до 3 раз
props.put("max.in.flight.requests.per.connection", "1");
// ↑ Важно для сохранения порядка

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = 
        new ProducerRecord<>("my-topic", "msg-" + i);
    
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            // Пересылаем
            System.out.println("Error, will retry: " + exception);
        } else {
            System.out.println("Message sent to " + metadata.partition());
        }
    });
}

producer.close();

На уровне потребителя (Consumer)

// Потребитель должен идемпотентно обработать дубли
@KafkaListener(topics = "my-topic")
public void handleMessage(Message message) {
    // Кэш обработанных сообщений
    if (processedCache.contains(message.getId())) {
        System.out.println("Duplicate, skipping");
        return;  // Уже обработан
    }
    
    // Обработать
    process(message);
    processedCache.add(message.getId());
}

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

Преимущества:

• Сообщения не потеряны
• Гарантирована доставка
• Хорошее соотношение надёжности и производительности

Проблемы:

• Возможны дубли — потребитель должен быть идемпотентным
• Задержка выше (ждём подтверждений)
• Нельзя гарантировать порядок при retries

Когда использовать

// Подходит для большинства cases:
// - Email отправка (дубль письма — понят пользователю)
// - Notifications
// - Бизнес-события
// - Order placement (если идемпотентно реализовано)

public class OrderService {
    public void createOrder(Order order) {
        orderRepository.save(order);
        
        // At-Least-Once гарантия
        kafkaProducer.send(
            new ProducerRecord<>("orders", order.getId().toString(), order),
            (metadata, exception) -> {
                if (exception != null) {
                    logger.error("Failed to send order event", exception);
                }
            }
        );
    }
}

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

Сообщение доставляется ровно 1 раз. Не потеряно и не продублировано.

Как это работает

Производитель → отправка → Брокер (с ID)
    ↓                           ↓
  сохранить ID          отклонить дубли
  (в БД или кэш)        с тем же ID

Реализация в Kafka

// Kafka с idempotence и transactional producer
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");

// Exactly-Once
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
props.put("enable.idempotence", true);  // ← Ключевое свойство
props.put("transactional.id", "my-transactional-producer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Инициализировать транзакцию
producer.initTransactions();

try {
    producer.beginTransaction();
    
    // Отправить несколько сообщений как атомарная операция
    for (int i = 0; i < 10; i++) {
        producer.send(
            new ProducerRecord<>("my-topic", "msg-" + i),
            (metadata, exception) -> {
                if (exception != null) {
                    throw new RuntimeException(exception);
                }
            }
        );
    }
    
    // Если мы здесь — всё ОК
    producer.commitTransaction();
} catch (Exception e) {
    // Если ошибка — откатить всё
    producer.abortTransaction();
    throw e;
}

Со стороны потребителя

// Потребитель должен использовать transactional обработку
@KafkaListener(topics = "my-topic")
public void handleMessage(Message message) {
    // Использовать database transaction
    transactionTemplate.execute((status) -> {
        // Step 1: Обработать
        processMessage(message);
        
        // Step 2: Сохранить в БД
        resultRepository.save(result);
        
        // Step 3: Сохранить offset сообщения
        offsetRepository.save(new ProcessedOffset(
            message.getTopic(),
            message.getPartition(),
            message.getOffset()
        ));
        
        // Если всё ОК — commit БД транзакции
        // Если ошибка — rollback
        return null;
    });
}

Более сложный пример: Exactly-Once с БД

public class PaymentProcessor {
    
    @Autowired
    private PaymentRepository paymentRepository;
    
    @Autowired
    private KafkaProducer<String, PaymentEvent> producer;
    
    // Обработка платежа с гарантией exactly-once
    public void processPayment(String paymentId) {
        // Check: платёж уже обработан?
        Payment existing = paymentRepository.findById(paymentId).orElse(null);
        if (existing != null && existing.isProcessed()) {
            System.out.println("Payment already processed");
            return;  // Идемпотентно
        }
        
        // Начать транзакцию
        producer.beginTransaction();
        try {
            // Step 1: Обновить платёж в БД
            Payment payment = paymentRepository.findById(paymentId)
                .orElseThrow();
            payment.setProcessed(true);
            payment.setProcessedAt(LocalDateTime.now());
            paymentRepository.save(payment);
            
            // Step 2: Отправить событие
            producer.send(new ProducerRecord<>(
                "payment-processed",
                paymentId,
                new PaymentEvent(paymentId, "PROCESSED")
            ));
            
            // Step 3: Коммит
            producer.commitTransaction();
            
        } catch (Exception e) {
            producer.abortTransaction();
            throw e;
        }
    }
}

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

Преимущества:

• Нет дублей — каждое сообщение обработано ровно один раз
• Нет потерь — гарантирована доставка
• Идеально для критичных операций

Проблемы:

• Самая низкая производительность — много проверок и синхронизаций
• Сложная реализация — требует поддержки транзакций в системе
• Стоимость — больше ресурсов на брокере и потребителе

Когда использовать

// Подходит только для критичных операций:
// - Платёжные транзакции
// - Финансовые операции
// - Юридические события
// - Данные, которые нельзя потерять или продублировать

public class FinancialTransactionService {
    
    // Используем Exactly-Once
    @Transactional
    public void transferMoney(Long fromAccount, Long toAccount, BigDecimal amount) {
        // Step 1: Дебет со счёта
        accountRepository.debit(fromAccount, amount);
        
        // Step 2: Кредит на счёт
        accountRepository.credit(toAccount, amount);
        
        // Step 3: Отправить событие с гарантией
        kafkaProducer.send(
            new ProducerRecord<>(
                "financial-transactions",
                UUID.randomUUID().toString(),
                new FinancialEvent(
                    fromAccount, toAccount, amount
                )
            )
        );
    }
}

Сравнение гарантий

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

Выбери At-Most-Once если:

• Потеря данных приемлема
• Нужна максимальная производительность
• Пример: логирование, аналитика

props.put("acks", "0");
props.put("retries", "0");

Выбери At-Least-Once если:

• Нужна гарантия доставки
• Потребитель может идемпотентно обрабатывать дубли
• Это подходит для большинства приложений

props.put("acks", "all");
props.put("retries", "3");
props.put("max.in.flight.requests.per.connection", "1");

Выбери Exactly-Once если:

• Нельзя потерять или продублировать сообщение
• Это финансовые или критичные операции
• Готов к снижению производительности

props.put("enable.idempotence", true);
props.put("transactional.id", "unique-id");

Идемпотентность

Ключ к Exactly-Once — идемпотентная обработка:

// НЕ идемпотентно
public void incrementBalance(Long accountId, BigDecimal amount) {
    Account acc = accountRepository.findById(accountId).orElseThrow();
    acc.setBalance(acc.getBalance().add(amount));
    accountRepository.save(acc);
    // Дубль = неправильный баланс!
}

// Идемпотентно
public void updateBalance(Long accountId, BigDecimal newBalance) {
    Account acc = accountRepository.findById(accountId).orElseThrow();
    acc.setBalance(newBalance);  // Устанавливаем, не добавляем
    accountRepository.save(acc);
    // Дубль = результат тот же
}