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.
        }
    }
}

# application.yml — manual commit, no auto-commit
spring:
  kafka:
    consumer:
      enable-auto-commit: false
      auto-offset-reset: earliest
    listener:
      ack-mode: manual

Dead Letter Topic

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
    var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
        (record, ex) -> new TopicPartition(record.topic() + ".dlt", record.partition()));

    // 3 retries with backoff, then publish to <topic>.dlt instead of blocking
    // the partition forever on a poison message.
    return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3));
}

Topic Naming and Partition Keys

Anti-Patterns

• Auto-commit with business logic in the listener — a crash between poll and processing silently drops the message. Use manual Acknowledgment.
• No idempotency check — Kafka is at-least-once; the same record WILL be redelivered after a rebalance or retry. Every consumer must tolerate duplicates.
• JSON without a schema — fine for prototyping, but breaks silently on producer/consumer version drift in production. Prefer Avro/Protobuf with Schema Registry per rules/infra.md.
• One giant topic for every event type — makes consumer group scaling and retention policy impossible to tune per event. Split by domain/entity.

Related

• Rule: infra.md — Kafka/Flink conventions this skill implements
• Rule: java-springboot.md — @KafkaListener/KafkaTemplate usage