At-Least-Once Delivery

// Safe at-least-once consumer — commit AFTER processing
Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);  // critical

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

    for (ConsumerRecord<String, String> r : records) {
        processOrder(r.value());  // DB write, side effects
    }

    // Only commit after ALL records in this batch are processed
    // If we crash before this line, these records are re-processed on restart
    consumer.commitSync();
}

// Scenario: crash after processOrder() but before commitSync()
// → Kafka re-delivers the same records on restart
// → processOrder() runs again → DUPLICATE

@KafkaListener(topics = "orders", groupId = "order-processor")
@Transactional
public void onOrder(ConsumerRecord<String, OrderEvent> record) {

    // Build a unique key from Kafka coordinates
    String eventKey = record.topic() + "-" + record.partition() + "-" + record.offset();

    // Deduplication check — same DB transaction as the business operation
    if (processedEventRepo.existsByKey(eventKey)) {
        log.info("Skipping duplicate: {}", eventKey);
        return;
    }

    // Business logic
    orderService.createOrder(record.value());

    // Mark as processed — atomic with the business write
    processedEventRepo.save(new ProcessedEvent(eventKey, Instant.now()));
    // Both writes are in the same @Transactional boundary
    // If anything fails, both roll back — the record will be reprocessed safely
}

// Producer-side deduplication: enable idempotent producer
Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
// Automatically sets: acks=all, retries=MAX_VALUE, max.in.flight.requests=5

// Without idempotence:
// 1. Producer sends record [PID=-, seq=-] → broker writes it → ACK lost
// 2. Producer retries → broker writes it again → DUPLICATE

// With idempotence:
// 1. Producer sends [PID=42, seq=0] → broker writes, stores seq
// 2. Producer retries [PID=42, seq=0] → broker sees same seq → DISCARDS

// Application-level correlation ID as an alternative
@Service
public class OrderEventPublisher {
    public void publish(Order order) {
        OrderEvent event = OrderEvent.builder()
            .eventId(order.getId().toString())  // stable, deterministic ID
            .orderId(order.getId())
            .build();
        // Consumer can use eventId to deduplicate
        kafka.send("orders", order.getId().toString(), event);
    }
}