Kafka and Database Transactions Together

// With spring.kafka.producer.transaction-id-prefix set, Spring Boot gives the listener container a
// KafkaTransactionManager. The container starts a Kafka transaction before the listener runs and,
// when the listener returns, sends the consumed offsets to that transaction and commits it.
// @Transactional adds the database's own transaction, committed first.

@Service
public class ExactlyOnceProcessor {

    // "dstm" is the name of your database transaction manager bean
    @Transactional("dstm")
    @KafkaListener(topics = "payments.payment.completed")
    public void processPayment(PaymentEvent event) {
        // Idempotent database write: this method can run again if the Kafka commit fails
        Order order = orderRepo.findById(event.getOrderId()).orElseThrow();
        order.setStatus("PAID");
        orderRepo.save(order);

        // Sent inside the container's Kafka transaction
        kafkaTemplate.send("orders.order.paid",
            event.getOrderId(), new OrderPaidEvent(order));
    }
}