Transactional Kafka Producer

spring:
  kafka:
    producer:
      transaction-id-prefix: order-service-tx-
      # When this is set, Spring Boot configures a KafkaTransactionManager
    consumer:
      properties:
        isolation.level: read_committed   # consumers then skip aborted transactional records

@Service
@Slf4j
public class TransactionalOrderService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    // Option 1: @Transactional with the KafkaTransactionManager bean
    @Transactional("kafkaTransactionManager")
    public void processOrderWithEvents(Order order) {
        // Both sends are in one Kafka transaction: they commit or abort together
        kafkaTemplate.send("orders.order.created",
            order.getId(), new OrderCreatedEvent(order));
        kafkaTemplate.send("inventory.reservation.requested",
            order.getId(), new ReservationRequest(order));
        // If an exception leaves this method, both records are aborted
    }

    // Option 2: explicit transaction around a callback
    public void processWithExplicitTx(Order order) {
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("orders.order.created", order.getId(), new OrderCreatedEvent(order));
            operations.send("payments.payment.requested", order.getId(), new PaymentRequest(order));
            return true;   // normal exit commits the transaction
        });
    }
}