#Step 09 — 트랜잭션

// [9-3] transaction-id-prefix 를 켠 상태에서 그냥 send
template.send("orders", "ORD-0001", OrderCreated.of(1));   // ← 폭발

KafkaTemplate<String, Object> t = new KafkaTemplate<>(pf);
t.setAllowNonTransactional(true);     // 트랜잭션 없으면 그냥 보낸다

// [9-4] 두 토픽에 원자적으로 발행 — 예외를 던져 abort 시켜 봅니다
template.executeInTransaction(t -> {
    t.send("orders",   "ORD-0002", OrderCreated.of(2));
    t.send("payments", "ORD-0002", PaymentRequested.from(OrderCreated.of(2)));
    throw new IllegalStateException("[9-4] 결제 한도 초과 — 트랜잭션을 abort 합니다");
});

// [9-6] 컨테이너가 레코드마다 트랜잭션을 시작하도록 설정
ConcurrentKafkaListenerContainerFactory<String, Object> f = new ConcurrentKafkaListenerContainerFactory<>();
f.setConsumerFactory(cf);
f.getContainerProperties().setTransactionManager(ktm);              // Spring Kafka 3.1.x (이 코스)
// f.getContainerProperties().setKafkaAwareTransactionManager(ktm); // Spring Kafka 3.2+

@KafkaListener(topics = "orders", groupId = "s09-cpp", containerFactory = "cppFactory")
void process(ConsumerRecord<String, OrderCreated> rec) {
    log.info("[9-6] consume {}@{} {}", rec.topic(), rec.offset(), rec.key());
    template.send("payments", rec.key(), PaymentRequested.from(rec.value()));   // 트랜잭션 안
}

@Transactional("dbTxManager")
public void createOrder(OrderCreated order) {
    jdbc.update("INSERT INTO orders(order_id, customer_id, amount, status) VALUES (?,?,?,?)",
                order.orderId(), order.customerId(), order.amount(), "CREATED");
    kafkaTemplate.send("orders", order.orderId(), order);
    if (order.amount().intValue() > 12000) {
        throw new IllegalStateException("[9-7] 한도 초과 — DB 롤백");
    }
}

TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
    @Override public void afterCommit() {
        kafkaTemplate.send("orders", order.orderId(), order);   // DB 커밋 확정 후 발행
    }
});

// Spring Framework 5.3 / Spring Kafka 2.7 부터 deprecated. 시작은 좌→우, 커밋은 우→좌.
new ChainedKafkaTransactionManager<>(kafkaTxManager, dataSourceTxManager);

// [9-11] 같은 transactional.id 를 쓰는 두 프로듀서 (둘 다 tx-fence-0)
KafkaTemplate<String, Object> a = new KafkaTemplate<>(newFactory("tx-fence-"));
KafkaTemplate<String, Object> b = new KafkaTemplate<>(newFactory("tx-fence-"));

a.executeInTransaction(t -> t.send("orders", "FENCE-1", OrderCreated.of(101)));  // A: epoch 0
b.executeInTransaction(t -> t.send("orders", "FENCE-2", OrderCreated.of(102)));  // B: epoch 1 — A 를 펜싱
a.executeInTransaction(t -> t.send("orders", "FENCE-3", OrderCreated.of(103)));  // A: 사망

package com.example.order.step09;

/*
 * ============================================================================
 * Step 09 — 트랜잭션 : Practice
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step09/Practice.java
 *
 * application.yml 에 아래 두 줄을 추가한 상태에서 실행합니다.
 *
 *   spring.kafka.producer.transaction-id-prefix: tx-order-
 *   spring.kafka.consumer.properties.isolation.level: read_committed
 *
 * 실행 (보조 프로필을 하나만 함께 켭니다. 예제끼리 서로 간섭합니다)
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-nontx'
 *       → [9-3] ★ 트랜잭션 밖 send 가 IllegalStateException 으로 죽는 것
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-exec'
 *       → [9-4] executeInTransaction 으로 orders + payments 원자 발행 / abort
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-isolation'
 *       → [9-5] ★ read_uncommitted 가 abort 된 메시지를 받는 것 (두 그룹 나란히)
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-cpp'
 *       → [9-6] consume-process-produce. sendOffsetsToTransaction 로그 확인
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-ghost'
 *       → [9-7A] ★★ 유령 이벤트. DB 는 롤백, Kafka 에는 남음
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-lost'
 *       → [9-7B] afterCommit 발행. DB 는 있는데 이벤트가 없음
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-bench'
 *       → [9-10] 처리량 실측. -Dbench.total=100000 으로 줄여도 배수는 같음
 *   ./gradlew bootRun --args='--spring.profiles.active=step09,step09-fence'
 *       → [9-11] ★ 같은 transactional.id 두 개 → ProducerFencedException
 *
 * ★ 9-7 은 실행 전 주문 행을 지우세요. 두 번째 실행부터는 PK 중복으로
 *   INSERT 가 먼저 터져 시나리오가 성립하지 않습니다.
 *
 *   mq 'DELETE FROM orders WHERE order_id IN ("ORD-0013","ORD-0014")'
 *
 * ★ 9-5 / 9-6 은 실행 전 오프셋을 리셋하세요 (앱을 먼저 종료할 것).
 *
 *   kcg --group s09-uncommitted --topic orders --reset-offsets --to-earliest --execute
 *   kcg --group s09-committed   --topic orders --reset-offsets --to-earliest --execute
 *   kcg --group s09-cpp         --topic orders --reset-offsets --to-earliest --execute
 *
 * 실행 중에 확인할 CLI
 *   alias mq='docker exec -i learn-kafka-mysql mysql -ulearner -plearn1234 orderdb -t -e'
 *   mq 'SELECT * FROM orders'                                    ← ★ 9-7 의 절반
 *   kcc --topic orders   --from-beginning                        ← read_uncommitted (기본)
 *   kcc --topic orders   --from-beginning --isolation-level read_committed
 *   kcc --topic payments --from-beginning --isolation-level read_committed
 *   kcg --describe --group s09-cpp
 *
 * 로그 레벨 (트랜잭션 흐름을 보려면 필수)
 *   logging.level.org.springframework.kafka.transaction: DEBUG
 *   logging.level.org.apache.kafka.clients.producer.internals.TransactionManager: INFO
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;

public final class Practice {

    private Practice() { }

    static final String ORDERS = "orders";
    static final String PAYMENTS = "payments";
    static final String BOOTSTRAP = "127.0.0.1:9092";

    /** payments 토픽에 실을 이벤트. orders 와 다른 토픽이어야 원자성 실습이 됩니다. */
    public record PaymentRequested(String orderId, BigDecimal amount, Instant requestedAt) {
        public static PaymentRequested from(OrderCreated o) {
            return new PaymentRequested(o.orderId(), o.amount(), o.createdAt());
        }
    }

    // ========================================================================
    // [9-0] 공통 빈
    // ========================================================================
    @Configuration
    @Profile("step09")
    static class CommonConfig {

        /**
         * ⚠️ [9-2] 트랜잭션 매니저는 반드시 이름을 지어 두고, @Transactional 에서 이름으로 지정합니다.
         * transaction-id-prefix 를 켜면 KafkaTransactionManager 가 자동 등록되는데,
         * Boot 의 JPA 자동 설정은 @ConditionalOnMissingBean(TransactionManager.class) 조건이라
         * 등록 순서에 따라 DB 트랜잭션 매니저가 아예 안 만들어질 수 있습니다.
         */
        @Bean("dbTxManager")
        DataSourceTransactionManager dbTxManager(DataSource ds) {
            return new DataSourceTransactionManager(ds);
        }

        @Bean
        JdbcTemplate jdbcTemplate(DataSource ds) {
            return new JdbcTemplate(ds);
        }

        /**
         * [9-7] "평범한" 비트랜잭션 프로듀서.
         * application.yml 의 transaction-id-prefix 와 무관한 팩토리를 따로 만들어,
         * 대부분의 프로젝트가 쓰는 상태를 한 앱 안에서 재현합니다.
         */
        @Bean("plainTemplate")
        KafkaTemplate<String, Object> plainTemplate() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
            // transactional.id 없음 → 트랜잭션 개념 자체가 없는 프로듀서
            return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
        }
    }

    // ========================================================================
    // [9-3] 트랜잭션 밖 send — IllegalStateException
    // ========================================================================
    @Component
    @Profile("step09-nontx")
    static class NonTxSendDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(NonTxSendDemo.class);
        private final KafkaTemplate<String, Object> template;

        NonTxSendDemo(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            log.info("[9-3] 트랜잭션 밖에서 send 를 시도합니다");
            try {
                template.send(ORDERS, "ORD-0001", OrderCreated.of(1));
                log.warn("[9-3] 예외가 안 났습니다. transaction-id-prefix 가 켜져 있는지 확인하세요");
            } catch (IllegalStateException e) {
                // 기대 메시지:
                //   No transaction is in process; possible solutions: run the template operation
                //   within the scope of a template.executeInTransaction() operation, start a
                //   transaction with @Transactional before invoking the template method, run in a
                //   transaction started by a listener container when consuming a record
                log.error("[9-3] {}", e.getMessage());
            }

            // 해결책 ① — executeInTransaction 으로 감싼다 (권장)
            template.executeInTransaction(t -> t.send(ORDERS, "ORD-0001", OrderCreated.of(1)));
            log.info("[9-3] executeInTransaction 으로 감싸니 정상 발행되었습니다");

            // 해결책 ② — allowNonTransactional. 예외만 꺼질 뿐 원자성은 없습니다.
            // template.setAllowNonTransactional(true);
            // template.send(ORDERS, "ORD-0001", OrderCreated.of(1));
        }
    }

    // ========================================================================
    // [9-4] executeInTransaction — 두 토픽 원자 발행
    // ========================================================================
    @Component
    @Profile("step09-exec")
    static class ExecInTxDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(ExecInTxDemo.class);
        private final KafkaTemplate<String, Object> template;

        ExecInTxDemo(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            // (1) 커밋되는 트랜잭션 — orders 와 payments 양쪽에 남습니다
            OrderCreated ok = OrderCreated.of(1);
            template.executeInTransaction(t -> {
                t.send(ORDERS, ok.orderId(), ok);
                t.send(PAYMENTS, ok.orderId(), PaymentRequested.from(ok));
                return null;
            });
            log.info("[9-4] 커밋 완료: {}", ok.orderId());

            // (2) abort 되는 트랜잭션 — 양쪽 어디에도 안 남습니다
            OrderCreated bad = OrderCreated.of(2);
            log.info("[9-4] 트랜잭션 시작");
            try {
                template.executeInTransaction(t -> {
                    log.info("[9-4] send orders   {}", bad.orderId());
                    t.send(ORDERS, bad.orderId(), bad);
                    log.info("[9-4] send payments {}", bad.orderId());
                    t.send(PAYMENTS, bad.orderId(), PaymentRequested.from(bad));
                    throw new IllegalStateException("[9-4] 결제 한도 초과 — 트랜잭션을 abort 합니다");
                });
            } catch (RuntimeException e) {
                log.error("[9-4] abort 되었습니다: {}", e.getMessage());
            }

            log.info("[9-4] 확인: kcc --topic orders   --from-beginning --isolation-level read_committed");
            log.info("[9-4] 확인: kcc --topic payments --from-beginning --isolation-level read_committed");
            log.info("[9-4] 두 토픽 모두 {} 가 없어야 정답입니다", bad.orderId());
        }
    }

    // ========================================================================
    // [9-5] isolation.level — read_uncommitted vs read_committed
    // ========================================================================
    @Configuration
    @Profile("step09-isolation")
    static class IsolationConfig {

        private Map<String, Object> base(String isolation) {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation);   // 문자열이어야 합니다
            return props;
        }

        @Bean("uncommittedFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> uncommittedFactory(
                ConsumerFactory<String, Object> defaults) {
            return factory(defaults, "read_uncommitted");
        }

        @Bean("committedFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> committedFactory(
                ConsumerFactory<String, Object> defaults) {
            return factory(defaults, "read_committed");
        }

        private ConcurrentKafkaListenerContainerFactory<String, Object> factory(
                ConsumerFactory<String, Object> defaults, String isolation) {
            Map<String, Object> props = new HashMap<>(defaults.getConfigurationProperties());
            props.putAll(base(isolation));
            ConcurrentKafkaListenerContainerFactory<String, Object> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
            f.setConcurrency(1);
            return f;
        }
    }

    @Component
    @Profile("step09-isolation")
    static class IsolationDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(IsolationDemo.class);
        private final KafkaTemplate<String, Object> template;

        IsolationDemo(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            send(11, true);    // 커밋
            send(12, false);   // abort  ← 이 한 건이 이 절의 주인공
            send(13, true);    // 커밋
            Thread.sleep(3000);
            log.info("[9-5] read_uncommitted 는 ORD-0012 를 받고, read_committed 는 안 받습니다");
        }

        private void send(int seq, boolean commit) {
            OrderCreated e = OrderCreated.of(seq);
            try {
                template.executeInTransaction(t -> {
                    t.send(ORDERS, e.orderId(), e);
                    if (!commit) {
                        throw new IllegalStateException("[9-5] 의도적 abort: " + e.orderId());
                    }
                    return null;
                });
                log.info("[9-5] 발행(커밋) {}", e.orderId());
            } catch (RuntimeException ex) {
                log.warn("[9-5] 발행(abort) {}", e.orderId());
            }
        }

        @KafkaListener(topics = ORDERS, groupId = "s09-uncommitted",
                       containerFactory = "uncommittedFactory")
        void readUncommitted(OrderCreated e) {
            log.info("[9-5] read_uncommitted 수신: {}", e.orderId());
        }

        @KafkaListener(topics = ORDERS, groupId = "s09-committed",
                       containerFactory = "committedFactory")
        void readCommitted(OrderCreated e) {
            log.info("[9-5] read_committed   수신: {}", e.orderId());
        }
    }

    // ========================================================================
    // [9-6] consume-process-produce
    // ========================================================================
    @Configuration
    @Profile("step09-cpp")
    static class CppConfig {

        @Bean("cppFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> cppFactory(
                ConsumerFactory<String, Object> cf,
                KafkaTransactionManager<String, Object> ktm) {

            ConcurrentKafkaListenerContainerFactory<String, Object> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(cf);
            f.setConcurrency(1);

            // Spring Kafka 3.1.x (이 코스)
            f.getContainerProperties().setTransactionManager(ktm);

            // ⚠️ Spring Kafka 3.2+ : 위 메서드가 deprecated 되었습니다. 아래로 교체하세요.
            //    새 시그니처는 KafkaAwareTransactionManager 만 받으므로,
            //    DataSourceTransactionManager 를 잘못 넣는 실수가 컴파일 에러가 됩니다.
            // f.getContainerProperties().setKafkaAwareTransactionManager(ktm);

            return f;
        }
    }

    @Component
    @Profile("step09-cpp")
    static class CppListener implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(CppListener.class);
        private final KafkaTemplate<String, Object> template;

        CppListener(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            // 소비할 재료를 먼저 넣어 둡니다
            template.executeInTransaction(t -> {
                for (int i = 1; i <= 6; i++) {
                    OrderCreated e = OrderCreated.of(i);
                    t.send(ORDERS, e.orderId(), e);
                }
                return null;
            });
            log.info("[9-6] 재료 6건 발행 완료");
        }

        /**
         * 컨테이너가 레코드 소비 시점에 Kafka 트랜잭션을 시작해 두었으므로,
         * 아래 send 는 9-3 의 IllegalStateException 없이 그 트랜잭션에 올라탑니다.
         * 커밋 직전 컨테이너가 sendOffsetsToTransaction 을 자동 호출합니다.
         */
        @KafkaListener(topics = ORDERS, groupId = "s09-cpp", containerFactory = "cppFactory")
        void process(ConsumerRecord<String, OrderCreated> rec) {
            log.info("[9-6] consume {}@{} {}", rec.topic(), rec.offset(), rec.key());
            template.send(PAYMENTS, rec.key(), PaymentRequested.from(rec.value()));

            // ORD-0005 에서만 실패시켜, 발행과 오프셋 커밋이 함께 무효가 되는 것을 봅니다
            if ("ORD-0005".equals(rec.key())) {
                throw new IllegalStateException("[9-6] 처리 실패 — 트랜잭션 롤백");
            }
        }
    }

    // ========================================================================
    // [9-7A] 유령 이벤트 — DB 는 롤백, Kafka 에는 남음
    // ========================================================================
    @Component
    @Profile("step09-ghost")
    static class GhostOrderService {

        private static final Logger log = LoggerFactory.getLogger(GhostOrderService.class);
        private final JdbcTemplate jdbc;
        private final KafkaTemplate<String, Object> plain;

        GhostOrderService(JdbcTemplate jdbc, @Qualifier("plainTemplate") KafkaTemplate<String, Object> plain) {
            this.jdbc = jdbc;
            this.plain = plain;
        }

        /**
         * 흔히 보는 코드입니다. 그리고 원자적이지 않습니다.
         * plain 은 비트랜잭션 프로듀서이므로 send 는 DB 커밋과 무관하게 즉시 나갑니다.
         */
        @Transactional("dbTxManager")
        public void createOrder(OrderCreated order) {
            log.info("[9-7A] INSERT orders {} amount={}", order.orderId(), order.amount());
            jdbc.update("INSERT INTO orders(order_id, customer_id, amount, status) VALUES (?,?,?,?)",
                    order.orderId(), order.customerId(), order.amount(), "CREATED");

            log.info("[9-7A] kafka send → {}", ORDERS);
            plain.send(ORDERS, order.orderId(), order);

            if (order.amount().intValue() > 12_000) {
                throw new IllegalStateException("[9-7] 한도 초과 — DB 롤백");
            }
        }
    }

    @Component
    @Profile("step09-ghost")
    static class GhostDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(GhostDemo.class);
        private final GhostOrderService service;
        private final JdbcTemplate jdbc;

        GhostDemo(GhostOrderService service, JdbcTemplate jdbc) {
            this.service = service;
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            OrderCreated order = OrderCreated.of(13);   // amount = 13000 → 한도 초과
            try {
                service.createOrder(order);
            } catch (RuntimeException e) {
                log.error("[9-7A] 롤백: {}", e.getMessage());
            }
            Thread.sleep(1000);

            Integer rows = jdbc.queryForObject(
                    "SELECT COUNT(*) FROM orders WHERE order_id = ?", Integer.class, order.orderId());
            log.info("[9-7A] DB 의 {} 행 수 = {}  (0 이면 롤백된 것)", order.orderId(), rows);
            log.info("[9-7A] 이제 토픽을 보세요:");
            log.info("[9-7A]   kcc --topic orders --from-beginning --isolation-level read_committed | grep {}",
                    order.orderId());
            log.info("[9-7A] DB 에 없는 주문의 이벤트가 토픽에 있으면 = 유령 이벤트 재현 성공");
        }
    }

    // ========================================================================
    // [9-7B] 이벤트 유실 — afterCommit 으로 순서를 뒤집었을 때
    // ========================================================================
    @Component
    @Profile("step09-lost")
    static class LostOrderService {

        private static final Logger log = LoggerFactory.getLogger(LostOrderService.class);
        private final JdbcTemplate jdbc;
        private final KafkaTemplate<String, Object> plain;

        LostOrderService(JdbcTemplate jdbc, @Qualifier("plainTemplate") KafkaTemplate<String, Object> plain) {
            this.jdbc = jdbc;
            this.plain = plain;
        }

        /**
         * @Transactional 메서드 "안"의 마지막 줄에 send 를 둬도 커밋 전입니다.
         * 진짜로 순서를 뒤집으려면 afterCommit 콜백이 필요합니다.
         * 그러면 유령 이벤트는 사라지지만, 이번엔 발행 실패 = 유실이 됩니다.
         */
        @Transactional("dbTxManager")
        public void createOrder(OrderCreated order, boolean brokerDown) {
            log.info("[9-7B] INSERT orders {}", order.orderId());
            jdbc.update("INSERT INTO orders(order_id, customer_id, amount, status) VALUES (?,?,?,?)",
                    order.orderId(), order.customerId(), order.amount(), "CREATED");

            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    log.info("[9-7B] afterCommit — kafka send 시도");
                    try {
                        // brokerDown 이면 존재하지 않는 토픽으로 보내 발행 실패를 흉내 냅니다.
                        // 실제 재현은 docker stop learn-kafka 후 실행하는 쪽이 정확합니다.
                        String topic = brokerDown ? "orders-broker-down-simulation" : ORDERS;
                        plain.send(topic, order.orderId(), order).get();
                        log.info("[9-7B] 발행 성공");
                    } catch (Exception e) {
                        log.error("[9-7B] 발행 실패: {}", e.getMessage());
                        log.error("[9-7B] DB 는 이미 커밋되었습니다. 되돌릴 수 없습니다 = 이벤트 유실");
                    }
                }
            });
        }
    }

    @Component
    @Profile("step09-lost")
    static class LostDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(LostDemo.class);
        private final LostOrderService service;
        private final JdbcTemplate jdbc;

        LostDemo(LostOrderService service, JdbcTemplate jdbc) {
            this.service = service;
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated order = OrderCreated.of(14);
            service.createOrder(order, true);

            Integer rows = jdbc.queryForObject(
                    "SELECT COUNT(*) FROM orders WHERE order_id = ?", Integer.class, order.orderId());
            log.info("[9-7B] DB 의 {} 행 수 = {}  (1 이면 커밋된 것)", order.orderId(), rows);
            log.info("[9-7B] 주문은 있는데 이벤트는 없습니다. 재고도 안 빠지고 알림도 안 갑니다");
            log.info("[9-7B] 다만 DB 가 진실의 원천이므로 나중에 복구할 수 있습니다 → Outbox (9-9)");
        }
    }

    // ========================================================================
    // [9-10] 트랜잭션의 비용
    // ========================================================================
    @Component
    @Profile("step09-bench")
    static class ThroughputBench implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(ThroughputBench.class);
        private final KafkaTemplate<String, Object> tx;
        private final KafkaTemplate<String, Object> plain;
        private final int total;

        ThroughputBench(KafkaTemplate<String, Object> tx,
                        @Qualifier("plainTemplate") KafkaTemplate<String, Object> plain,
                        @Value("${bench.total:1000000}") int total) {
            this.tx = tx;
            this.plain = plain;
            this.total = total;
        }

        @Override
        public void run(ApplicationArguments args) {
            measureNonTx();
            measureTx(1);
            measureTx(100);
            measureTx(1000);
        }

        private void measureNonTx() {
            long t0 = System.nanoTime();
            for (int i = 0; i < total; i++) {
                plain.send(ORDERS, "BENCH-" + i, OrderCreated.of(i));
            }
            plain.flush();
            report("non-tx        ", t0);
        }

        private void measureTx(int batch) {
            long t0 = System.nanoTime();
            for (int i = 0; i < total; i += batch) {
                final int from = i;
                tx.executeInTransaction(t -> {
                    for (int j = from; j < Math.min(from + batch, total); j++) {
                        t.send(ORDERS, "BENCH-" + j, OrderCreated.of(j));
                    }
                    return null;
                });
            }
            report("tx batch=" + batch, t0);
        }

        private void report(String label, long t0) {
            long ms = (System.nanoTime() - t0) / 1_000_000;
            long perSec = ms == 0 ? 0 : total * 1000L / ms;
            log.info("[9-10] {}  {}건  {}ms  {} msg/s", label, total, ms, perSec);
        }
    }

    // ========================================================================
    // [9-11] transactional.id 충돌 → ProducerFencedException
    // ========================================================================
    @Component
    @Profile("step09-fence")
    static class FencingDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(FencingDemo.class);

        /** 두 "인스턴스"를 흉내 냅니다. 같은 접두사 → 같은 transactional.id (tx-fence-0). */
        private DefaultKafkaProducerFactory<String, Object> newFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            DefaultKafkaProducerFactory<String, Object> pf = new DefaultKafkaProducerFactory<>(props);
            pf.setTransactionIdPrefix("tx-fence-");
            return pf;
        }

        @Override
        public void run(ApplicationArguments args) {
            DefaultKafkaProducerFactory<String, Object> pfA = newFactory();
            DefaultKafkaProducerFactory<String, Object> pfB = newFactory();
            KafkaTemplate<String, Object> a = new KafkaTemplate<>(pfA);
            KafkaTemplate<String, Object> b = new KafkaTemplate<>(pfB);

            try {
                a.executeInTransaction(t -> t.send(ORDERS, "FENCE-1", OrderCreated.of(101)));
                log.info("[9-11] A 발행 성공 FENCE-1   (epoch 0)");

                b.executeInTransaction(t -> t.send(ORDERS, "FENCE-2", OrderCreated.of(102)));
                log.info("[9-11] B 발행 성공 FENCE-2   (epoch 1 — A 를 펜싱했습니다)");

                a.executeInTransaction(t -> t.send(ORDERS, "FENCE-3", OrderCreated.of(103)));
                log.warn("[9-11] A 가 살아남았습니다. 두 팩토리의 접두사가 실제로 같은지 확인하세요");
            } catch (Exception e) {
                log.error("[9-11] A 사망: {}: {}", e.getClass().getName(), e.getMessage());
                log.error("[9-11] 운영에서는 파드 하나가 계속 재시작하는 모습으로 나타납니다");
            } finally {
                // ⚠️ 정리하지 않으면 미완료 트랜잭션이 남아
                //    다음 실습의 read_committed 컨슈머를 transaction.timeout.ms 만큼 막습니다.
                pfA.destroy();
                pfB.destroy();
            }

            log.info("[9-11] 해결: transaction-id-prefix: tx-order-${{HOSTNAME:local}}-  + StatefulSet");
        }
    }
}

package com.example.order.step09;

/*
 * ============================================================================
 * Step 09 — 트랜잭션 : Exercise (문제 6개)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step09/Exercise.java
 *
 * application.yml 전제
 *   spring.kafka.producer.transaction-id-prefix: tx-order-
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step09ex,step09ex-1'   … -6
 *
 * 확인 명령
 *   alias mq='docker exec -i learn-kafka-mysql mysql -ulearner -plearn1234 orderdb -t -e'
 *   kcc --topic orders   --from-beginning                                   ← read_uncommitted
 *   kcc --topic orders   --from-beginning --isolation-level read_committed   ← ★ 문제 2 는 이걸로
 *   kcc --topic payments --from-beginning --isolation-level read_committed
 *   kcg --describe --group s09-ex-cpp
 *
 * ⚠️ 문제 4 는 실행 전에 반드시:
 *   mq 'DELETE FROM orders WHERE order_id="ORD-9004"'
 *   두 번째 실행부터는 PK 중복 예외가 send 보다 먼저 터져서 아무 일도 안 일어납니다.
 *
 * 문제 순서대로 푸세요. 2번이 만든 abort 메시지가 있어야 3번 비교가 성립합니다.
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;

public final class Exercise {

    private Exercise() { }

    static final String ORDERS = "orders";
    static final String PAYMENTS = "payments";
    static final String BOOTSTRAP = "127.0.0.1:9092";

    public record PaymentRequested(String orderId, BigDecimal amount, Instant requestedAt) {
        public static PaymentRequested from(OrderCreated o) {
            return new PaymentRequested(o.orderId(), o.amount(), o.createdAt());
        }
    }

    // ------------------------------------------------------------------------
    // 공통 빈 (수정하지 마세요)
    // ------------------------------------------------------------------------
    @Configuration
    @Profile("step09ex")
    static class CommonConfig {

        @Bean("dbTxManager")
        DataSourceTransactionManager dbTxManager(DataSource ds) {
            return new DataSourceTransactionManager(ds);
        }

        @Bean
        JdbcTemplate jdbcTemplate(DataSource ds) {
            return new JdbcTemplate(ds);
        }

        /** transactional.id 가 없는 "평범한" 프로듀서. 문제 4 에서 씁니다. */
        @Bean("plainTemplate")
        KafkaTemplate<String, Object> plainTemplate() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
        }
    }

    // ========================================================================
    // 문제 1. 트랜잭션 밖 send 의 예외를 확인하고 해결하기
    //
    //   요구사항
    //   (a) transaction-id-prefix 가 켜진 상태에서 template.send(...) 를 그대로 호출하고,
    //       발생하는 IllegalStateException 의 메시지 "전문"을 ERROR 로 남길 것.
    //   (b) 그다음 같은 발행을 예외 없이 성공시킬 것. 단 allowNonTransactional 은 쓰지 말 것.
    //   (c) 왜 allowNonTransactional 이 답이 아닌지 주석 한 줄로 적을 것.
    //
    //   확인: 로그에 "No transaction is in process; possible solutions:" 가 찍힐 것
    // ========================================================================
    @Component
    @Profile("step09ex-1")
    static class Ex1 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex1.class);
        private final KafkaTemplate<String, Object> template;

        Ex1(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated e = OrderCreated.of(9001);

            // 여기에 작성: (a) 트랜잭션 밖 send → 예외 메시지 전문 로깅

            // 여기에 작성: (b) executeInTransaction 으로 감싸 성공시키기

            // 여기에 작성: (c) allowNonTransactional 이 답이 아닌 이유 (주석)
        }
    }

    // ========================================================================
    // 문제 2. 두 토픽 원자 발행과 abort
    //
    //   요구사항
    //   (a) executeInTransaction 안에서 orders 와 payments 에 각각 ORD-9002 를 발행할 것.
    //   (b) 세 번째 줄에서 RuntimeException 을 던져 abort 시킬 것.
    //   (c) 예외를 잡아 로그로 남길 것.
    //
    //   확인 (★ 반드시 --isolation-level read_committed 를 붙여서):
    //     kcc --topic orders   --from-beginning --isolation-level read_committed | grep ORD-9002
    //     kcc --topic payments --from-beginning --isolation-level read_committed | grep ORD-9002
    //   둘 다 아무것도 안 나와야 정답입니다.
    // ========================================================================
    @Component
    @Profile("step09ex-2")
    static class Ex2 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex2.class);
        private final KafkaTemplate<String, Object> template;

        Ex2(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated e = OrderCreated.of(9002);

            // 여기에 작성: (a)(b)(c)
        }
    }

    // ========================================================================
    // 문제 3. 격리 수준이 다른 두 컨슈머로 표 완성하기
    //
    //   요구사항
    //   (a) isolation.level 만 다른 컨테이너 팩토리 두 개를 만들 것
    //       - exUncommittedFactory : read_uncommitted
    //       - exCommittedFactory   : read_committed
    //   (b) 각각을 쓰는 @KafkaListener 두 개를 groupId 를 달리해 붙일 것
    //       - s09-ex-uncommitted / s09-ex-committed
    //   (c) ORD-9031(커밋) / ORD-9032(abort) / ORD-9033(커밋) 을 발행할 것
    //
    //   관측 기록 (실행 후 직접 채우세요):
    //     | 그룹                  | ORD-9031 | ORD-9032(abort) | ORD-9033 |
    //     |----------------------|----------|-----------------|----------|
    //     | s09-ex-uncommitted   |          |                 |          |
    //     | s09-ex-committed     |          |                 |          |
    //
    //   확인: read_uncommitted 쪽만 ORD-9032 를 수신할 것
    // ========================================================================
    @Configuration
    @Profile("step09ex-3")
    static class Ex3Config {

        // 여기에 작성: (a) 두 팩토리 빈
        // 힌트: ConsumerConfig.ISOLATION_LEVEL_CONFIG 의 값은 문자열입니다.
        //       defaults.getConfigurationProperties() 를 복사한 뒤 덮어쓰면 편합니다.
    }

    @Component
    @Profile("step09ex-3")
    static class Ex3 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex3.class);
        private final KafkaTemplate<String, Object> template;

        Ex3(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            // 여기에 작성: (c) 9031 커밋 / 9032 abort / 9033 커밋 발행
            Thread.sleep(3000);
        }

        // 여기에 작성: (b) 리스너 두 개
    }

    // ========================================================================
    // 문제 4. ★ 유령 이벤트 재현
    //
    //   ⚠️ 실행 전: mq 'DELETE FROM orders WHERE order_id="ORD-9004"'
    //
    //   요구사항
    //   (a) @Transactional("dbTxManager") 메서드 안에서
    //       ① orders 테이블에 INSERT
    //       ② plainTemplate 으로 orders 토픽에 발행
    //       ③ RuntimeException 을 던져 DB 를 롤백
    //   (b) 롤백 후 DB 행 수와 토픽 내용을 각각 확인할 것
    //
    //   ★ @Transactional 은 프록시로 동작하므로, 같은 클래스 안에서 자기 메서드를 호출하면
    //     트랜잭션이 안 걸립니다. 반드시 별도 빈(Ex4Service)에 두고 주입해서 호출하세요.
    //
    //   관측 기록 (실행 후 직접 채우세요):
    //     mq 'SELECT * FROM orders WHERE order_id="ORD-9004"'  →
    //     kcc --topic orders --from-beginning --isolation-level read_committed | grep ORD-9004  →
    //
    //   확인: DB 는 Empty set 인데 토픽에는 ORD-9004 가 1건 있을 것
    // ========================================================================
    @Component
    @Profile("step09ex-4")
    static class Ex4Service {

        private static final Logger log = LoggerFactory.getLogger(Ex4Service.class);
        private final JdbcTemplate jdbc;
        private final KafkaTemplate<String, Object> plain;

        Ex4Service(JdbcTemplate jdbc, @Qualifier("plainTemplate") KafkaTemplate<String, Object> plain) {
            this.jdbc = jdbc;
            this.plain = plain;
        }

        // 여기에 작성: (a) @Transactional("dbTxManager") createOrder(OrderCreated)
    }

    @Component
    @Profile("step09ex-4")
    static class Ex4 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex4.class);
        private final Ex4Service service;
        private final JdbcTemplate jdbc;

        Ex4(Ex4Service service, JdbcTemplate jdbc) {
            this.service = service;
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            OrderCreated order = new OrderCreated("ORD-9004", 1004, "SKU-001", 2,
                    new BigDecimal(13_000), Instant.parse("2025-01-01T00:04:00Z"));

            // 여기에 작성: (b) service.createOrder(order) 호출 + 예외 처리 + DB 행 수 확인
        }
    }

    // ========================================================================
    // 문제 5. consume-process-produce 구성
    //
    //   요구사항
    //   (a) 리스너 컨테이너 팩토리에 KafkaTransactionManager 를 설정할 것 (exCppFactory)
    //   (b) orders 를 s09-ex-cpp 그룹으로 소비해 payments 로 발행하는 리스너를 만들 것
    //   (c) 재료로 ORD-9051 ~ ORD-9053 을 미리 발행해 둘 것
    //
    //   확인 방법
    //     logging.level.org.springframework.kafka.transaction=DEBUG 로 올린 뒤
    //     로그에서 "sendOffsetsToTransaction" 을 찾을 것.
    //     이 줄이 없으면 트랜잭션 매니저가 컨테이너에 안 붙은 것입니다.
    // ========================================================================
    @Configuration
    @Profile("step09ex-5")
    static class Ex5Config {

        // 여기에 작성: (a) exCppFactory 빈
        // 힌트: f.getContainerProperties().setTransactionManager(ktm)   ← 3.1.x
        //       Spring Kafka 3.2+ 라면 setKafkaAwareTransactionManager(ktm)
    }

    @Component
    @Profile("step09ex-5")
    static class Ex5 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex5.class);
        private final KafkaTemplate<String, Object> template;

        Ex5(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            // 여기에 작성: (c) 재료 3건 발행
        }

        // 여기에 작성: (b) @KafkaListener(groupId = "s09-ex-cpp", containerFactory = "exCppFactory")
    }

    // ========================================================================
    // 문제 6. ProducerFencedException 재현
    //
    //   요구사항
    //   (a) 같은 transactional.id 접두사("tx-ex-fence-")를 쓰는
    //       DefaultKafkaProducerFactory 두 개를 코드로 만들 것
    //   (b) A 로 한 번 발행 → B 로 한 번 발행 → A 로 다시 발행
    //   (c) A 가 받는 예외의 클래스명과 메시지를 로그로 남길 것
    //   (d) finally 에서 두 팩토리를 destroy() 할 것 (안 하면 다음 실습이 60초 막힙니다)
    //
    //   확인: 로그에 "ProducerId set to <n> with epoch 1" 이 찍히고,
    //         이어서 ProducerFencedException 이 날 것
    // ========================================================================
    @Component
    @Profile("step09ex-6")
    static class Ex6 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Ex6.class);

        @Override
        public void run(ApplicationArguments args) {
            // 여기에 작성: (a)(b)(c)(d)
        }
    }
}

package com.example.order.step09;

/*
 * ============================================================================
 * Step 09 — 트랜잭션 : Solution (정답 6개)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step09/Solution.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step09sol,step09sol-1'   … -6
 *
 * ⚠️ 정답 4 는 실행 전에:  mq 'DELETE FROM orders WHERE order_id="ORD-9004"'
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;

public final class Solution {

    private Solution() { }

    static final String ORDERS = "orders";
    static final String PAYMENTS = "payments";
    static final String BOOTSTRAP = "127.0.0.1:9092";

    public record PaymentRequested(String orderId, BigDecimal amount, Instant requestedAt) {
        public static PaymentRequested from(OrderCreated o) {
            return new PaymentRequested(o.orderId(), o.amount(), o.createdAt());
        }
    }

    @Configuration
    @Profile("step09sol")
    static class CommonConfig {

        @Bean("dbTxManager")
        DataSourceTransactionManager dbTxManager(DataSource ds) {
            return new DataSourceTransactionManager(ds);
        }

        @Bean
        JdbcTemplate jdbcTemplate(DataSource ds) {
            return new JdbcTemplate(ds);
        }

        @Bean("plainTemplate")
        KafkaTemplate<String, Object> plainTemplate() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
        }
    }

    // ========================================================================
    // 정답 1 — 트랜잭션 밖 send
    // ========================================================================
    /*
     * 답: executeInTransaction 으로 감싼다. allowNonTransactional 은 쓰지 않는다.
     *
     * 왜 예외가 나는가
     *   transaction-id-prefix 를 설정하면 KafkaTemplate 의 transactional 플래그가 true 가 됩니다.
     *   이 상태의 KafkaTemplate 은 doSend 직전에 "지금 진행 중인 트랜잭션이 있는가" 를 확인하고,
     *   없으면 프로듀서를 아예 내주지 않습니다. 그 결과가 IllegalStateException 입니다.
     *   진행 중인 트랜잭션으로 인정되는 것은 세 가지뿐입니다.
     *     ① executeInTransaction 콜백 안
     *     ② @Transactional 로 시작된 트랜잭션 안
     *     ③ 리스너 컨테이너가 레코드 소비 시 시작한 트랜잭션 안
     *   예외 메시지가 이 세 가지를 그대로 나열해 줍니다. 메시지를 끝까지 읽는 것이 가장 빠른 해결입니다.
     *
     * 왜 allowNonTransactional 은 답이 아닌가
     *   이 옵션은 "트랜잭션이 없으면 트랜잭션 없이 보낸다" 입니다. 즉 예외만 사라지고
     *   원자성은 여전히 없습니다. 더 나쁜 것은 동작이 호출 문맥에 의존하게 된다는 점입니다.
     *   같은 send 한 줄이, 나중에 누가 그 메서드를 @Transactional 안으로 옮기면
     *   갑자기 트랜잭션 안에서 돌기 시작합니다. 로그에도 흔적이 없어서 사후 추적이 불가능합니다.
     *   레거시를 단계적으로 이관하는 기간에만 켜고, 끝나면 반드시 끄십시오.
     *
     * 확인 로그
     *   ERROR ... : No transaction is in process; possible solutions: run the template operation
     *               within the scope of a template.executeInTransaction() operation, ...
     */
    @Component
    @Profile("step09sol-1")
    static class Sol1 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol1.class);
        private final KafkaTemplate<String, Object> template;

        Sol1(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated e = OrderCreated.of(9001);

            // (a) 트랜잭션 밖 send
            try {
                template.send(ORDERS, e.orderId(), e);
                log.warn("정답1(a) 예외가 안 났습니다. transaction-id-prefix 설정을 확인하세요");
            } catch (IllegalStateException ex) {
                log.error("정답1(a) {}", ex.getMessage());
            }

            // (b) executeInTransaction 으로 해결
            template.executeInTransaction(t -> t.send(ORDERS, e.orderId(), e));
            log.info("정답1(b) executeInTransaction 으로 발행 성공: {}", e.orderId());

            // (c) allowNonTransactional 은 예외만 끄고 원자성은 주지 않으며,
            //     트랜잭션 유무가 호출 문맥에 따라 조용히 바뀌게 만들기 때문에 답이 아닙니다.
        }
    }

    // ========================================================================
    // 정답 2 — 두 토픽 원자 발행과 abort
    // ========================================================================
    /*
     * 답: executeInTransaction 콜백 안에서 두 번 send 하고, 그 뒤에 예외를 던진다.
     *
     * 핵심은 코드가 아니라 검증 명령입니다.
     *   kcc --topic orders --from-beginning                      ← ORD-9002 가 "보입니다"
     *   kcc --topic orders --from-beginning --isolation-level read_committed  ← 안 보입니다
     *   이 문제를 틀리는 대부분의 이유가 옵션 없이 확인하고
     *   "트랜잭션이 안 걸렸다" 고 오진하는 것입니다. 기본값이 read_uncommitted 이기 때문입니다.
     *
     * 왜 두 토픽이어야 하는가
     *   orders 는 파티션 3, payments 는 파티션 1 입니다. 서로 다른 토픽, 서로 다른 파티션에
     *   걸친 쓰기가 통째로 무효가 되는 것을 봐야 "원자적 쓰기" 를 확인한 것입니다.
     *   한 토픽 한 파티션만으로는 멱등 프로듀서와 구분이 되지 않습니다.
     *
     * 콜백 안에서 하지 말 것
     *   send(...).get() 을 부르지 마세요. 커밋은 콜백이 끝난 뒤이므로 get() 으로 얻는 것은
     *   "브로커가 레코드를 받았다" 까지입니다. 커밋 여부가 아닙니다.
     *   게다가 매 건 get() 은 배치를 무력화해 처리량을 무너뜨립니다(Step 02).
     */
    @Component
    @Profile("step09sol-2")
    static class Sol2 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol2.class);
        private final KafkaTemplate<String, Object> template;

        Sol2(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated e = OrderCreated.of(9002);
            try {
                template.executeInTransaction(t -> {
                    t.send(ORDERS, e.orderId(), e);
                    t.send(PAYMENTS, e.orderId(), PaymentRequested.from(e));
                    throw new IllegalStateException("정답2 의도적 abort");
                });
            } catch (RuntimeException ex) {
                log.error("정답2 abort: {}", ex.getMessage());
            }
            log.info("정답2 확인: kcc --topic orders   --from-beginning --isolation-level read_committed | grep {}", e.orderId());
            log.info("정답2 확인: kcc --topic payments --from-beginning --isolation-level read_committed | grep {}", e.orderId());
        }
    }

    // ========================================================================
    // 정답 3 — 격리 수준 비교
    // ========================================================================
    /*
     * 답: ConsumerConfig.ISOLATION_LEVEL_CONFIG 를 문자열 "read_uncommitted" / "read_committed" 로
     *     달리한 컨슈머 팩토리 두 개 + groupId 가 다른 리스너 두 개.
     *
     * 주의점 두 가지
     *   ① 값은 반드시 소문자 문자열입니다. "READ_COMMITTED" 나 enum 을 넣으면 기동 시
     *      ConfigException 이 납니다. 이건 그나마 즉시 드러나서 다행인 경우입니다.
     *   ② groupId 를 나누지 않으면 두 리스너가 같은 그룹에 들어가 파티션을 나눠 갖습니다.
     *      그러면 한쪽만 메시지를 받아 비교 자체가 성립하지 않습니다. 로그가 조용해서
     *      "read_committed 가 필터링했나?" 로 오해하기 딱 좋습니다.
     *
     * 관측 결과
     *   | 그룹                | ORD-9031 | ORD-9032(abort) | ORD-9033 |
     *   |--------------------|----------|-----------------|----------|
     *   | s09-ex-uncommitted |    수신   |    ★ 수신 ★     |   수신    |
     *   | s09-ex-committed   |    수신   |     안 받음      |   수신    |
     *
     * 이 표의 의미
     *   프로듀서 쪽 트랜잭션은 컨슈머가 read_committed 여야 비로소 의미를 가집니다.
     *   발행 팀과 소비 팀이 다르면, 발행 팀이 아무리 정성껏 트랜잭션을 도입해도
     *   소비 팀이 설정 한 줄을 안 바꾸면 효과가 0 입니다.
     *   트랜잭션 도입 계획의 첫 항목은 코드가 아니라 "컨슈머 목록 작성" 입니다.
     *
     * read_committed 의 대가
     *   컨슈머는 LSO(Last Stable Offset) 까지만 읽습니다. 진행 중인 트랜잭션이 하나 있으면
     *   그 뒤에 커밋된 메시지가 아무리 쌓여도 볼 수 없습니다.
     *   죽은 프로듀서 하나가 transaction.timeout.ms(기본 60초) 동안 파티션을 막습니다.
     */
    @Configuration
    @Profile("step09sol-3")
    static class Sol3Config {

        @Bean("solUncommittedFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> solUncommittedFactory(
                ConsumerFactory<String, Object> defaults) {
            return factory(defaults, "read_uncommitted");
        }

        @Bean("solCommittedFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> solCommittedFactory(
                ConsumerFactory<String, Object> defaults) {
            return factory(defaults, "read_committed");
        }

        private ConcurrentKafkaListenerContainerFactory<String, Object> factory(
                ConsumerFactory<String, Object> defaults, String isolation) {
            Map<String, Object> props = new HashMap<>(defaults.getConfigurationProperties());
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation);

            ConcurrentKafkaListenerContainerFactory<String, Object> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
            f.setConcurrency(1);
            return f;
        }
    }

    @Component
    @Profile("step09sol-3")
    static class Sol3 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol3.class);
        private final KafkaTemplate<String, Object> template;

        Sol3(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            publish(9031, true);
            publish(9032, false);
            publish(9033, true);
            Thread.sleep(3000);
        }

        private void publish(int seq, boolean commit) {
            OrderCreated e = OrderCreated.of(seq);
            try {
                template.executeInTransaction(t -> {
                    t.send(ORDERS, e.orderId(), e);
                    if (!commit) {
                        throw new IllegalStateException("정답3 의도적 abort " + e.orderId());
                    }
                    return null;
                });
                log.info("정답3 발행(커밋) {}", e.orderId());
            } catch (RuntimeException ex) {
                log.warn("정답3 발행(abort) {}", e.orderId());
            }
        }

        @KafkaListener(topics = ORDERS, groupId = "s09-ex-uncommitted",
                       containerFactory = "solUncommittedFactory")
        void uncommitted(OrderCreated e) {
            log.info("정답3 read_uncommitted 수신: {}", e.orderId());
        }

        @KafkaListener(topics = ORDERS, groupId = "s09-ex-committed",
                       containerFactory = "solCommittedFactory")
        void committed(OrderCreated e) {
            log.info("정답3 read_committed   수신: {}", e.orderId());
        }
    }

    // ========================================================================
    // 정답 4 — ★ 유령 이벤트
    // ========================================================================
    /*
     * 답: @Transactional("dbTxManager") 메서드 안에서 INSERT → plain.send → 예외.
     *     코드는 세 줄이고, 정답의 본체는 관측 기록 두 줄입니다.
     *
     *   mq 'SELECT * FROM orders WHERE order_id="ORD-9004"'
     *     → Empty set
     *   kcc --topic orders --from-beginning --isolation-level read_committed | grep ORD-9004
     *     → {"orderId":"ORD-9004", ...}   ← 있습니다
     *
     *   존재하지 않는 주문의 이벤트가 이미 발행되었습니다. 하류에서는 재고가 차감되고
     *   알림이 나가고 정산 배치가 이 주문을 집계합니다. DB 에는 없는 주문에 대해서.
     *
     * 왜 send 를 마지막 줄로 옮겨도 안 되는가
     *   @Transactional 메서드 "안" 에는 커밋 시점이 없습니다. 커밋은 메서드가 정상 리턴한 뒤
     *   프록시가 수행합니다. 그러므로 메서드 안 어디에 두든 순서는 항상 send → commit 입니다.
     *   진짜로 뒤집으려면 TransactionSynchronization.afterCommit 콜백이 필요합니다.
     *   그런데 그렇게 하면 이번엔 "DB 는 커밋됐는데 Kafka 발행이 실패" 하는 유실이 생깁니다.
     *   즉 문제가 해결되는 것이 아니라 트레이드오프가 이동할 뿐입니다.
     *
     * 트랜잭션 프로듀서를 쓰면 나아지는가
     *   조금 나아지고, 사라지지는 않습니다. transaction-id-prefix 가 있으면 KafkaTemplate 은
     *   진행 중인 DB 트랜잭션에 동기화로 참여해 DB 커밋 후 Kafka 를 커밋합니다.
     *   그래도 ① DB 커밋 성공 직후 Kafka 커밋 실패 시 유실이 남고,
     *          ② 컨슈머가 read_uncommitted(기본값) 면 abort 된 메시지를 읽어
     *             유령 이벤트가 그대로 부활합니다.
     *   두 시스템에 걸친 원자적 커밋은 2PC 가 필요한데 Kafka 는 XA 리소스가 아닙니다.
     *
     * 결론
     *   Outbox 로 가십시오. 업무 데이터와 이벤트를 같은 DB 트랜잭션에 넣으면
     *   원자성이 필요한 지점이 DB 하나로 줄어듭니다. Step 13 에서 구현합니다.
     */
    @Component
    @Profile("step09sol-4")
    static class Sol4Service {

        private static final Logger log = LoggerFactory.getLogger(Sol4Service.class);
        private final JdbcTemplate jdbc;
        private final KafkaTemplate<String, Object> plain;

        Sol4Service(JdbcTemplate jdbc, @Qualifier("plainTemplate") KafkaTemplate<String, Object> plain) {
            this.jdbc = jdbc;
            this.plain = plain;
        }

        @Transactional("dbTxManager")
        public void createOrder(OrderCreated order) {
            log.info("정답4 INSERT orders {}", order.orderId());
            jdbc.update("INSERT INTO orders(order_id, customer_id, amount, status) VALUES (?,?,?,?)",
                    order.orderId(), order.customerId(), order.amount(), "CREATED");

            log.info("정답4 kafka send → {} (비트랜잭션 프로듀서이므로 즉시 나갑니다)", ORDERS);
            plain.send(ORDERS, order.orderId(), order);

            throw new IllegalStateException("정답4 한도 초과 — DB 롤백");
        }
    }

    @Component
    @Profile("step09sol-4")
    static class Sol4 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol4.class);
        private final Sol4Service service;
        private final JdbcTemplate jdbc;

        Sol4(Sol4Service service, JdbcTemplate jdbc) {
            this.service = service;
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            OrderCreated order = new OrderCreated("ORD-9004", 1004, "SKU-001", 2,
                    new BigDecimal(13_000), Instant.parse("2025-01-01T00:04:00Z"));
            try {
                service.createOrder(order);
            } catch (RuntimeException e) {
                log.error("정답4 롤백: {}", e.getMessage());
            }
            Thread.sleep(1000);

            Integer rows = jdbc.queryForObject(
                    "SELECT COUNT(*) FROM orders WHERE order_id = ?", Integer.class, order.orderId());
            log.info("정답4 DB 행 수 = {}  (0 이면 롤백됨)", rows);
            log.info("정답4 kcc --topic orders --from-beginning --isolation-level read_committed | grep ORD-9004");
            log.info("정답4 DB 0건 + 토픽 1건 = 유령 이벤트");
        }
    }

    // ========================================================================
    // 정답 5 — consume-process-produce
    // ========================================================================
    /*
     * 답: ContainerProperties 에 KafkaTransactionManager 를 설정한다.
     *
     *   Spring Kafka 3.1.x  : f.getContainerProperties().setTransactionManager(ktm);
     *   Spring Kafka 3.2 +  : f.getContainerProperties().setKafkaAwareTransactionManager(ktm);
     *
     * 왜 3.2 에서 메서드가 바뀌었는가
     *   기존 setTransactionManager 는 PlatformTransactionManager 를 받습니다. 그래서
     *   DataSourceTransactionManager 를 넣어도 컴파일되고 기동도 됩니다. 그러면 컨테이너는
     *   DB 트랜잭션만 시작할 뿐 소비 오프셋을 Kafka 트랜잭션에 넣지 않습니다.
     *   여러분은 consume-process-produce 를 켰다고 믿지만 실제로는 안 켜져 있고,
     *   로그에 아무 차이도 없습니다. 이 코스가 말하는 "조용히 잘못 동작하는" 전형입니다.
     *   새 메서드는 KafkaAwareTransactionManager 만 받으므로 이 실수가 컴파일 에러가 됩니다.
     *
     * 켜졌는지 확인하는 방법
     *   logging.level.org.springframework.kafka.transaction=DEBUG 로 올리고
     *   로그에서 아래 줄을 찾으세요.
     *     sendOffsetsToTransaction: {orders-1=OffsetAndMetadata{offset=1}} group=s09-ex-cpp
     *   이 줄이 없으면 트랜잭션 매니저가 컨테이너에 안 붙은 것입니다.
     *   여러분이 코드로 부른 적 없는 sendOffsetsToTransaction 이 호출되는 것이
     *   "발행 + 오프셋 커밋이 하나의 트랜잭션" 이라는 유일한 증거입니다.
     *
     * 부수 효과 두 가지
     *   ① AckMode 설정이 무시됩니다. 오프셋은 항상 트랜잭션 커밋 시점에 나갑니다.
     *   ② 트랜잭션 경계가 poll 단위입니다. 레코드 리스너면 사실상 1건당 1 트랜잭션이라
     *      9-10 표의 840 msg/s 구간입니다. 처리량이 필요하면 배치 리스너로 바꾸세요.
     *      대신 배치 중 한 건이 실패하면 배치 전체가 재처리됩니다(Step 06).
     */
    @Configuration
    @Profile("step09sol-5")
    static class Sol5Config {

        @Bean("solCppFactory")
        ConcurrentKafkaListenerContainerFactory<String, Object> solCppFactory(
                ConsumerFactory<String, Object> cf,
                KafkaTransactionManager<String, Object> ktm) {

            ConcurrentKafkaListenerContainerFactory<String, Object> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(cf);
            f.setConcurrency(1);
            f.getContainerProperties().setTransactionManager(ktm);        // 3.1.x
            // f.getContainerProperties().setKafkaAwareTransactionManager(ktm);   // 3.2+
            return f;
        }
    }

    @Component
    @Profile("step09sol-5")
    static class Sol5 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol5.class);
        private final KafkaTemplate<String, Object> template;

        Sol5(KafkaTemplate<String, Object> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            template.executeInTransaction(t -> {
                for (int seq = 9051; seq <= 9053; seq++) {
                    OrderCreated e = OrderCreated.of(seq);
                    t.send(ORDERS, e.orderId(), e);
                }
                return null;
            });
            log.info("정답5 재료 3건 발행 완료");
        }

        @KafkaListener(topics = ORDERS, groupId = "s09-ex-cpp", containerFactory = "solCppFactory")
        void process(ConsumerRecord<String, OrderCreated> rec) {
            log.info("정답5 consume {}@{} {}", rec.topic(), rec.offset(), rec.key());
            template.send(PAYMENTS, rec.key(), PaymentRequested.from(rec.value()));
        }
    }

    // ========================================================================
    // 정답 6 — ProducerFencedException
    // ========================================================================
    /*
     * 답: 같은 접두사를 가진 DefaultKafkaProducerFactory 두 개를 만들고 번갈아 발행한다.
     *
     * 왜 이것이 "파드 2개" 와 같은 상황인가
     *   transactional.id 는 "접두사 + 카운터" 로 만들어지는데, 이 카운터는
     *   DefaultKafkaProducerFactory 인스턴스가 JVM 안에서 매기는 로컬 값입니다.
     *   다른 프로세스와 조율되지 않습니다. 그래서 같은 애플리케이션을 두 인스턴스 띄우면
     *   양쪽 다 tx-order-0 을 쓰고, 파드를 3개로 늘리면 셋 다 tx-order-0 입니다.
     *   팩토리 두 개를 한 JVM 에 만드는 것은 이 상황의 최소 재현입니다.
     *
     * 무슨 일이 일어나는가
     *   B 가 initTransactions 를 부르는 순간 코디네이터가 그 transactional.id 의 epoch 를
     *   0 에서 1 로 올립니다. 이때부터 epoch 0 인 A 의 요청은 전부 거절됩니다.
     *     ProducerId set to 2000 with epoch 0     ← A
     *     ProducerId set to 2000 with epoch 1     ← B (PID 는 같고 epoch 만 다릅니다)
     *     ProducerFencedException: There is a newer producer with the same transactionalId ...
     *   이것이 펜싱이고, 원래 목적은 죽은 줄 알았던 인스턴스가 되살아나
     *   옛 트랜잭션을 커밋하는 좀비 시나리오를 막는 것입니다.
     *
     * 운영에서의 증상
     *   파드 2개를 띄우면 하나는 정상, 하나는 계속 ProducerFencedException 으로 재시작합니다.
     *   스케일아웃할수록 나빠집니다. 예외는 명확히 찍히지만 원인이 "인스턴스 수" 라는 것을
     *   연결짓기까지 시간이 걸립니다.
     *
     * 해결과 그 해결의 새 문제
     *   transaction-id-prefix: tx-order-${HOSTNAME:local}-
     *   Kubernetes 에서 HOSTNAME 은 파드 이름이므로 인스턴스 간 충돌은 사라집니다.
     *   그런데 Deployment 의 파드 이름은 재시작마다 바뀝니다. 그러면 재시작 후
     *   신원이 달라져 펜싱이 동작하지 않고, 죽은 파드의 미완료 트랜잭션은
     *   transaction.timeout.ms 뒤 타임아웃 abort 될 때까지 남아
     *   read_committed 컨슈머를 그만큼 막습니다.
     *   그래서 트랜잭션 프로듀서를 쓰는 애플리케이션은 StatefulSet 이 정석입니다.
     *   파드 이름이 pod-0, pod-1 로 고정되어야 재시작 후에도 같은 신원으로 돌아옵니다.
     *
     * ⚠️ destroy() 를 잊지 마세요
     *   팩토리를 정리하지 않으면 미완료 트랜잭션이 남아 다음 실습의 read_committed
     *   컨슈머를 transaction.timeout.ms(기본 60초) 만큼 막습니다.
     */
    @Component
    @Profile("step09sol-6")
    static class Sol6 implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Sol6.class);

        private DefaultKafkaProducerFactory<String, Object> newFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            DefaultKafkaProducerFactory<String, Object> pf = new DefaultKafkaProducerFactory<>(props);
            pf.setTransactionIdPrefix("tx-ex-fence-");   // 두 팩토리가 같은 접두사 → 같은 id
            return pf;
        }

        @Override
        public void run(ApplicationArguments args) {
            DefaultKafkaProducerFactory<String, Object> pfA = newFactory();
            DefaultKafkaProducerFactory<String, Object> pfB = newFactory();
            KafkaTemplate<String, Object> a = new KafkaTemplate<>(pfA);
            KafkaTemplate<String, Object> b = new KafkaTemplate<>(pfB);

            try {
                a.executeInTransaction(t -> t.send(ORDERS, "FENCE-1", OrderCreated.of(9061)));
                log.info("정답6 A 발행 성공 FENCE-1 (epoch 0)");

                b.executeInTransaction(t -> t.send(ORDERS, "FENCE-2", OrderCreated.of(9062)));
                log.info("정답6 B 발행 성공 FENCE-2 (epoch 1 — A 를 펜싱)");

                a.executeInTransaction(t -> t.send(ORDERS, "FENCE-3", OrderCreated.of(9063)));
                log.warn("정답6 A 가 살아남았습니다. 두 팩토리의 접두사가 같은지 확인하세요");
            } catch (Exception e) {
                log.error("정답6 A 사망: {}: {}", e.getClass().getName(), e.getMessage());
            } finally {
                pfA.destroy();
                pfB.destroy();
            }
        }
    }
}