Exactly-Once with ChainedTransactionManager

@Configuration
public class ChainedTxConfig {

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTxManager(
        ProducerFactory<String, Object> pf
    ) {
        return new KafkaTransactionManager<>(pf);
    }

    // Chain Kafka + JPA transactions
    @Bean
    @Primary
    public ChainedKafkaTransactionManager<String, Object> chainedTxManager(
        KafkaTransactionManager<String, Object> ktm,
        JpaTransactionManager jtm
    ) {
        return new ChainedKafkaTransactionManager<>(ktm, jtm);
    }
}

@Service
public class ExactlyOnceProcessor {

    @Transactional("chainedTxManager")
    @KafkaListener(topics = "payments.payment.completed")
    public void processPayment(PaymentEvent event) {
        // Step 1: Write to DB (in DB transaction)
        Order order = orderRepo.findById(event.getOrderId()).orElseThrow();
        order.setStatus("PAID");
        orderRepo.save(order);  // DB write

        // Step 2: Produce to Kafka (in Kafka transaction)
        kafkaTemplate.send("orders.order.paid",
            event.getOrderId(), new OrderPaidEvent(order));

        // Both commit atomically — or both abort on exception
    }
}