トランザクション対応プロデューサーの実装

# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      # A unique ID prefix for the transactional producer
      transactional-id-prefix: my-app-tx-

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.producer.transactional-id-prefix}")
    private String transactionalIdPrefix;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all"); // Essential for transactions
        configProps.put(ProducerConfig.RETRIES_CONFIG, 0); // Kafka handles retries internally for transactions

        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(configProps);
        factory.setTransactionIdPrefix(transactionalIdPrefix); // Set the transactional ID prefix
        return factory;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaTransactionManagerConfig {

    @Bean
    public KafkaTransactionManager kafkaTransactionManager(ProducerFactory<String, String> producerFactory) {
        return new KafkaTransactionManager(producerFactory);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@Service
public class TransactionalProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public TransactionalProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Transactional
    public void sendGreeting(String user) {
        String message = "Hello, " + user + "!";
        kafkaTemplate.send("greetings-topic", user, message);
        System.out.println("Attempted to send: " + message);
    }

    // Main method for a runnable Spring Boot application
    @SpringBootApplication
    public static class DemoApplication {
        public static void main(String[] args) {
            // This would typically be run as a Spring Boot application
            // and the service called via a controller or runner.
            // For demonstration, we just show the structure.
            System.out.println("Run this as a Spring Boot app to use the service.");
            // SpringApplication.run(DemoApplication.class, args);
        }
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderProcessingService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public OrderProcessingService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Transactional
    public void processOrder(String orderId, String item) {
        // Send order creation event
        kafkaTemplate.send("order-created-topic", orderId, "Order " + orderId + " created for " + item);
        System.out.println("Sent order creation for: " + orderId);

        // Simulate a failure for demonstration
        if (orderId.equals("FAIL_ORDER")) {
            throw new RuntimeException("Simulated order processing failure!");
        }

        // Send inventory update event
        kafkaTemplate.send("inventory-update-topic", item, "Item " + item + " quantity reduced for order " + orderId);
        System.out.println("Sent inventory update for: " + item);

        System.out.println("Order " + orderId + " processed transactionally.");
    }

    // Main method for a runnable Spring Boot application
    public static void main(String[] args) {
        // This would typically be run as a Spring Boot application
        // and the service called via a controller or runner.
        System.out.println("Run this as a Spring Boot app to use the service.");
    }
}