kafka-patterns

Kafka Patterns (Spring Kafka)

Concrete implementations of the conventions in rules/infra.md: topic naming, manual commits, idempotent consumers, and dead letter topics.

When to Activate

• Adding a @KafkaListener consumer or a KafkaTemplate producer
• Designing a new topic or partition key
• Handling consumer failures without losing or duplicating messages
• Setting up a dead letter topic for a consumer group

Producer

@Service
public class OrderEventProducer {
    private final KafkaTemplate<String, OrderCancelledEvent> kafkaTemplate;

    public OrderEventProducer(KafkaTemplate<String, OrderCancelledEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishCancelled(Order order) {
        // Partition key = orderId keeps every event for one order in the same
        // partition, so a consumer sees them in order.
        String topic = "orders.order.cancelled";
        OrderCancelledEvent event = OrderCancelledEvent.from(order);

        kafkaTemplate.send(topic, order.getId().toString(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish {} for order {}", topic, order.getId(), ex);
                }
            });
    }
}

Consumer with Manual Commit + Idempotency

@Component
public class OrderCancelledConsumer {
    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public OrderCancelledConsumer(ProcessedEventRepository processedEvents, InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders.order.cancelled", groupId = "inventory-service")
    public void handle(ConsumerRecord<String, OrderCancelledEvent> record, Acknowledgment ack) {
        String eventId = record.key() + ":" + record.offset();

        // Idempotency guard: at-least-once delivery means this handler can run
        // more than once for the same record. Check-then-insert on a unique
        // constraint, not a plain SELECT, to survive concurrent redelivery.
        if (processedEvents.existsById(eventId)) {
            ack.acknowledge();
            return;
        }

        try {
            inventoryService.rollback(record.value().orderId());
            processedEvents.save(new ProcessedEvent(eventId, Instant.now()));
            ack.acknowledge();
        } catch (Exception ex) {
            log.error("Failed to process cancellation for order {}", record.value().orderId(), ex);
            // Do not acknowledge — the record will be redelivered. A retry
            // policy + dead letter topic (below) bounds how many times.
        }
    }
}