#Step 13 — 실전 패턴과 최종 프로젝트

public static final String HDR_MESSAGE_ID = "messageId";

ProducerRecord<String, OrderCreated> record =
        new ProducerRecord<>("orders.outbox", null, event.orderId(), event);
record.headers().add(HDR_MESSAGE_ID, ("OBX-" + row.id()).getBytes(StandardCharsets.UTF_8));
kafkaTemplate.send(record);

private static final String INSERT_MARK = """
        INSERT INTO processed_message (message_id, consumer_group)
        VALUES (?, ?)
        """;

boolean markProcessed(String messageId, String group) {
    try {
        jdbc.update(INSERT_MARK, group + ":" + messageId, group);
        return true;                     // 처음 보는 메시지
    } catch (DuplicateKeyException e) {
        return false;                    // 이미 처리함 → 스킵
    }
}

// ❌ 틀린 코드
Integer cnt = jdbc.queryForObject(
        "SELECT COUNT(*) FROM processed_message WHERE message_id = ?", Integer.class, key);
if (cnt != null && cnt > 0) {
    log.warn("duplicate — skipped");
    return;
}
deductStock(event);                                        // ← ①
jdbc.update("INSERT INTO processed_message ...", key, group);  // ← ②

// ❌ 틀린 코드 — 트랜잭션이 둘로 쪼개져 있다
@KafkaListener(topics = "orders.outbox", groupId = "s13-inventory")
public void onMessage(OrderCreated e, @Header("messageId") String messageId) {
    if (!markProcessed(messageId, GROUP)) return;   // 트랜잭션 #1 — 커밋됨
    inventoryService.deduct(e.sku(), e.quantity()); // 트랜잭션 #2 — 여기서 죽으면?
}

@KafkaListener(topics = "orders.outbox", groupId = "s13-inventory")
public void onMessage(OrderCreated e, @Header(HDR_MESSAGE_ID) String messageId) {
    inventoryTx.consumeOnce(messageId, GROUP, e);
}

// 별도 빈이어야 합니다 (자기 호출은 프록시를 안 탑니다)
@Transactional                                     // ← 이력 + 비즈니스가 한 트랜잭션
public void consumeOnce(String messageId, String group, OrderCreated e) {
    try {
        jdbc.update(INSERT_MARK, group + ":" + messageId, group);
    } catch (DuplicateKeyException dup) {
        log.warn("duplicate messageId={} group={} — skipped", messageId, group);
        return;                                    // 롤백 아님. 정상 종료
    }
    int updated = jdbc.update(
            "UPDATE inventory SET quantity = quantity - ? WHERE sku = ? AND quantity >= ?",
            e.quantity(), e.sku(), e.quantity());
    if (updated == 0) {
        throw new OutOfStockException(e.sku());    // ← 롤백 → 이력도 함께 사라짐
    }
}

@Scheduled(cron = "0 10 3 * * *")           // 매일 03:10
public void purge() {
    int total = 0, deleted;
    do {
        deleted = jdbc.update(                          // ★ LIMIT 로 쪼갠다
                "DELETE FROM processed_message WHERE processed_at < ? LIMIT 5000",
                Timestamp.from(Instant.now().minus(14, ChronoUnit.DAYS)));
        total += deleted;
    } while (deleted == 5000);
    log.info("purged {} rows from processed_message", total);
}

@Service
public class OrderService {

    @Transactional                                   // DataSourceTransactionManager
    public String createOrder(OrderCreated e) {
        jdbc.update("INSERT INTO orders (order_id, customer_id, amount, status)"
                        + " VALUES (?, ?, ?, 'CREATED')",
                e.orderId(), e.customerId(), e.amount());

        jdbc.update("INSERT INTO outbox_event (aggregate_id, event_type, payload)"
                        + " VALUES (?, 'OrderCreated', ?)",
                e.orderId(), toJson(e));

        log.info("order {} created (outbox staged)", e.orderId());
        return e.orderId();
    }
}

@Component
public class OutboxRelay {

    private static final String PICK = """
            SELECT id, aggregate_id, event_type, payload
            FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY id
            LIMIT 100
            FOR UPDATE SKIP LOCKED
            """;

    @Scheduled(fixedDelay = 500)
    @Transactional                                    // SELECT ~ UPDATE 가 한 트랜잭션
    public void relay() {
        List<OutboxRow> rows = jdbc.query(PICK, OUTBOX_ROW_MAPPER);
        if (rows.isEmpty()) return;

        for (OutboxRow row : rows) {
            OrderCreated event = fromJson(row.payload());
            ProducerRecord<String, OrderCreated> rec =
                    new ProducerRecord<>("orders.outbox", null, row.aggregateId(), event);
            rec.headers().add(HDR_MESSAGE_ID, ("OBX-" + row.id()).getBytes(UTF_8));
            rec.headers().add("eventType", row.eventType().getBytes(UTF_8));

            kafkaTemplate.send(rec).join();           // ★ 발행 확인까지 대기
            jdbc.update("UPDATE outbox_event SET published_at = NOW(3) WHERE id = ?", row.id());
        }
        log.info("relayed {} events (id {}..{})",
                rows.size(), rows.get(0).id(), rows.get(rows.size() - 1).id());
    }
}

// ❌ 절대 하면 안 되는 코드
@KafkaListener(topics = "orders.outbox", groupId = "s13-inventory")
public void onMessage(OrderCreated e) {
    executor.submit(() -> inventoryTx.consumeOnce(...));   // 즉시 리턴
}

@KafkaListener(topics = "orders.outbox", groupId = "s13-inventory", batch = "true")
public void onBatch(List<ConsumerRecord<String, OrderCreated>> records, Acknowledgment ack) {
    // 같은 키는 항상 같은 워커로 → 키 안에서 순서 유지
    Map<Integer, List<ConsumerRecord<String, OrderCreated>>> lanes = records.stream()
            .collect(Collectors.groupingBy(r -> Math.abs(r.key().hashCode()) % WORKERS));

    List<CompletableFuture<Void>> futures = lanes.values().stream()
            .map(lane -> CompletableFuture.runAsync(() -> lane.forEach(this::handleOne), executor))
            .toList();

    CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();  // ★ 반드시 대기
    ack.acknowledge();
}

package com.example.order.step13;

/*
 * ============================================================================
 * Step 13 — 실전 패턴과 최종 프로젝트 : Practice
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step13/Practice.java
 *
 * 사전 준비
 *   1) build.gradle 에 웹 스타터 추가 (13-8 의 REST 엔드포인트에 필요)
 *        implementation 'org.springframework.boot:spring-boot-starter-web'
 *   2) 토픽 생성
 *        kt --create --topic orders.outbox     --partitions 3 --replication-factor 1
 *        kt --create --topic orders.outbox.DLT --partitions 3 --replication-factor 1
 *   3) MySQL 별칭
 *        alias mq='docker exec -i learn-kafka-mysql mysql -ulearner -plearn1234 orderdb -t -e'
 *
 * 실행 (보조 프로필을 하나만 함께 켭니다. 예제끼리 서로 간섭합니다)
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,step13-idem'
 *       → [13-2] 멱등 컨슈머 + SELECT-then-INSERT 경합 재현(RaceDemo)
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,step13-split'
 *       → [13-3] ★함정★ 이력과 비즈니스가 다른 트랜잭션이면 메시지가 영구 스킵된다
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,step13-outbox'
 *       → [13-5] Transactional Outbox 전체 (OrderService + OutboxRelay)
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,step13-async'
 *       → [13-7] ★함정★ @Async 리스너가 처리 전에 커밋한다 vs 레인 워커
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,step13-order'
 *       → [13-8] 최종 프로젝트. REST + Outbox + 멱등 재고 + DLT + 알림 팬아웃
 *
 * 실행 전 DB 초기화 (매번 하세요. 안 하면 이력이 남아 전부 스킵됩니다)
 *   mq "TRUNCATE processed_message; TRUNCATE outbox_event; DELETE FROM orders;
 *       UPDATE inventory SET quantity = 1000;"
 *
 * 실행 중 확인할 CLI
 *   mq  "SELECT sku, quantity FROM inventory ORDER BY sku;"
 *   mq  "SELECT id, aggregate_id, created_at, published_at FROM outbox_event ORDER BY id;"
 *   mq  "SELECT message_id, consumer_group FROM processed_message ORDER BY processed_at;"
 *   kcg --describe --group s13-inventory
 *   kcc --topic orders.outbox --from-beginning --property print.key=true --property print.headers=true
 *
 * 커밋 시점을 눈으로 보려면 (13-7 에 필수)
 *   logging.level.org.springframework.kafka.listener=DEBUG
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.backoff.FixedBackOff;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public final class Practice {

    private Practice() {
    }

    /* =====================================================================
     * 공통 — 상수, SQL, 예외, 행 매퍼
     * ===================================================================== */

    /** 프로듀서가 만들어 실어 보내는 메시지 식별자 헤더. 13-2 의 후보 ③. */
    public static final String HDR_MESSAGE_ID = "messageId";
    public static final String HDR_EVENT_TYPE = "eventType";

    public static final String TOPIC_OUTBOX = "orders.outbox";
    public static final String TOPIC_OUTBOX_DLT = "orders.outbox.DLT";

    public static final String GROUP_INVENTORY = "s13-inventory";
    public static final String GROUP_NOTIFICATION = "s13-notification";

    /** 이 스텝의 모든 SQL. 먼저 읽고 나머지 코드를 보면 빠릅니다. */
    public static final class Sql {
        private Sql() {
        }

        /** 멱등 판정을 DB 제약에 위임한다. SELECT 가 없다는 점이 핵심. */
        public static final String INSERT_MARK = """
                INSERT INTO processed_message (message_id, consumer_group)
                VALUES (?, ?)
                """;

        /** ⚠️ 13-2 의 틀린 방식. 경합 재현용으로만 씁니다. */
        public static final String COUNT_MARK = """
                SELECT COUNT(*) FROM processed_message WHERE message_id = ?
                """;

        /**
         * quantity >= ? 조건이 재고 부족 검출의 전부입니다.
         * 조건에 안 맞으면 updated == 0 이 되고, 거기서 예외를 던져 트랜잭션을 롤백합니다.
         */
        public static final String DEDUCT = """
                UPDATE inventory SET quantity = quantity - ?
                WHERE sku = ? AND quantity >= ?
                """;

        public static final String INSERT_ORDER = """
                INSERT INTO orders (order_id, customer_id, amount, status)
                VALUES (?, ?, ?, 'CREATED')
                """;

        public static final String INSERT_OUTBOX = """
                INSERT INTO outbox_event (aggregate_id, event_type, payload)
                VALUES (?, ?, ?)
                """;

        /**
         * SKIP LOCKED — 다른 트랜잭션이 잠근 행은 기다리지 말고 건너뛴다.
         * 이것이 없으면 릴레이 인스턴스 2대가 같은 행을 집어 전량 중복 발행합니다.
         */
        public static final String PICK_OUTBOX = """
                SELECT id, aggregate_id, event_type, payload
                FROM outbox_event
                WHERE published_at IS NULL
                ORDER BY id
                LIMIT 100
                FOR UPDATE SKIP LOCKED
                """;

        public static final String MARK_PUBLISHED = """
                UPDATE outbox_event SET published_at = NOW(3) WHERE id = ?
                """;

        public static final String PURGE_MARKS = """
                DELETE FROM processed_message WHERE processed_at < ? LIMIT 5000
                """;

        public static final String OUTBOX_BACKLOG = """
                SELECT COUNT(*) FROM outbox_event WHERE published_at IS NULL
                """;
    }

    /** 재고 부족. 재시도해도 소용없지만, 재고 보충 가능성이 있으므로 재시도 대상으로 둡니다. */
    public static class OutOfStockException extends RuntimeException {
        public OutOfStockException(String sku, int need, int have) {
            super("out of stock %s need=%d have=%d".formatted(sku, need, have));
        }
    }

    public record OutboxRow(long id, String aggregateId, String eventType, String payload) {
    }

    public static final RowMapper<OutboxRow> OUTBOX_ROW_MAPPER = (rs, n) -> new OutboxRow(
            rs.getLong("id"),
            rs.getString("aggregate_id"),
            rs.getString("event_type"),
            rs.getString("payload"));

    /**
     * 멱등 키를 만드는 유일한 자리.
     * ⚠️ processed_message 의 PK 가 message_id 단독이므로 그룹명을 접두사로 붙입니다.
     *    안 붙이면 재고 그룹이 먼저 INSERT 한 순간 알림 그룹이 전부 스킵됩니다(13-2 함정).
     */
    public static String markKey(String group, String messageId) {
        return group + ":" + messageId;
    }

    /* =====================================================================
     * [13-2] 멱등 컨슈머 — 처리 이력 테이블
     * ===================================================================== */

    /**
     * ★ 이력 INSERT 와 재고 UPDATE 가 반드시 같은 트랜잭션이어야 합니다(13-3).
     * 그래서 리스너와 분리된 별도 빈으로 둡니다. 자기 호출(this.method())은 프록시를 안 타서
     * @Transactional 이 통째로 무시됩니다.
     */
    @Component
    public static class InventoryTx {

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

        private final JdbcTemplate jdbc;

        public InventoryTx(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @Transactional
        public void consumeOnce(String messageId, String group, OrderCreated e) {
            try {
                jdbc.update(Sql.INSERT_MARK, markKey(group, messageId), group);
                log.info("marked {} (tx started)", messageId);
            } catch (DuplicateKeyException dup) {
                // 롤백이 아니라 정상 종료입니다. 이미 처리한 메시지이므로 할 일이 없습니다.
                log.warn("duplicate messageId={} group={} — skipped", messageId, group);
                return;
            }

            int updated = jdbc.update(Sql.DEDUCT, e.quantity(), e.sku(), e.quantity());
            if (updated == 0) {
                Integer have = jdbc.queryForObject(
                        "SELECT quantity FROM inventory WHERE sku = ?", Integer.class, e.sku());
                log.warn("out of stock {} need={} have={} → rollback", e.sku(), e.quantity(), have);
                // ★ 여기서 던지면 위의 INSERT_MARK 도 함께 롤백됩니다.
                //   그래야 재시도가 정상 동작하고, 소진 후 DLT 로 갑니다.
                throw new OutOfStockException(e.sku(), e.quantity(), have == null ? -1 : have);
            }

            Integer remaining = jdbc.queryForObject(
                    "SELECT quantity FROM inventory WHERE sku = ?", Integer.class, e.sku());
            log.info("deducted {} -{} (remaining={})", e.sku(), e.quantity(), remaining);
        }
    }

    @Configuration
    @Profile("step13-idem")
    public static class IdempotentDemo {

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

        private final InventoryTx tx;

        public IdempotentDemo(InventoryTx tx) {
            this.tx = tx;
        }

        // [13-2] 멱등 컨슈머. 리스너는 메시지를 풀어 넘기기만 하고, 트랜잭션은 InventoryTx 가 엽니다.
        @KafkaListener(topics = TOPIC_OUTBOX, groupId = GROUP_INVENTORY)
        public void onMessage(OrderCreated e,
                              @Header(name = HDR_MESSAGE_ID, required = false) String messageId,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                              @Header(KafkaHeaders.OFFSET) long offset) {
            // 헤더가 없으면 멱등성이 성립하지 않습니다. 조용히 넘어가지 말고 실패시킵니다.
            if (messageId == null) {
                throw new IllegalStateException(
                        "missing %s header at %s@%d".formatted(HDR_MESSAGE_ID, topic, offset));
            }
            log.info("consume messageId={} order={} sku={} qty={}",
                    messageId, e.orderId(), e.sku(), e.quantity());
            tx.consumeOnce(messageId, GROUP_INVENTORY, e);
        }
    }

    /**
     * [13-2] ⚠️ SELECT-then-INSERT 의 경합 재현.
     * Kafka 를 전혀 쓰지 않습니다. 두 스레드를 CountDownLatch 로 동시에 출발시켜
     * "확인하고 하기" 패턴의 빈 구간만 순수하게 드러냅니다.
     */
    @Component
    @Profile("step13-idem")
    public static class RaceDemo implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(RaceDemo.class);
        private static final int ROUNDS = 100;

        private final JdbcTemplate jdbc;

        public RaceDemo(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            ExecutorService pool = Executors.newFixedThreadPool(2);
            AtomicInteger doubleDeduct = new AtomicInteger();

            for (int round = 1; round <= ROUNDS; round++) {
                String messageId = "RACE-" + round;
                jdbc.update("UPDATE inventory SET quantity = 1000 WHERE sku = 'SKU-002'");

                CountDownLatch start = new CountDownLatch(1);
                CountDownLatch done = new CountDownLatch(2);

                for (String tag : new String[]{"A", "B"}) {
                    pool.submit(() -> {
                        try {
                            start.await();
                            wrongIdempotentDeduct(tag, messageId);
                        } catch (Exception ignored) {
                            // 재현이 목적이므로 예외는 무시합니다
                        } finally {
                            done.countDown();
                        }
                    });
                }
                start.countDown();
                done.await();

                Integer qty = jdbc.queryForObject(
                        "SELECT quantity FROM inventory WHERE sku = 'SKU-002'", Integer.class);
                if (qty != null && qty < 997) {          // 정상이면 997 (1회 차감)
                    doubleDeduct.incrementAndGet();
                }
            }
            pool.shutdown();

            log.warn("SELECT-then-INSERT: {} / {} rounds double-deducted", doubleDeduct.get(), ROUNDS);
            log.warn("→ 실행할 때마다 숫자가 달라집니다. 그 사실 자체가 이 버그의 성질입니다.");
        }

        /** ❌ 틀린 코드. SELECT 와 INSERT 사이가 비어 있습니다. */
        private void wrongIdempotentDeduct(String tag, String messageId) {
            Integer cnt = jdbc.queryForObject(Sql.COUNT_MARK, Integer.class, messageId);
            log.info("[{}] select count={}", tag, cnt);
            if (cnt != null && cnt > 0) {
                log.warn("[{}] duplicate — skipped", tag);
                return;
            }
            jdbc.update(Sql.DEDUCT, 3, "SKU-002", 3);
            log.info("[{}] deducted SKU-002 -3", tag);
            try {
                jdbc.update(Sql.INSERT_MARK, messageId, "race");
                log.info("[{}] insert ok", tag);
            } catch (DuplicateKeyException dup) {
                log.warn("[{}] insert duplicate — but stock already deducted twice", tag);
            }
        }
    }

    /* =====================================================================
     * [13-3] ⚠️ 함정 — 이력과 비즈니스가 다른 트랜잭션이면 소용없다
     * ===================================================================== */

    @Component
    @Profile("step13-split")
    public static class SplitTxDemo {

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

        /** 이 오프셋을 처리할 때 차감 직전에 죽입니다. -1 이면 비활성. */
        private static final long CRASH_AT_OFFSET = 0L;

        private final JdbcTemplate jdbc;

        public SplitTxDemo(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @KafkaListener(topics = TOPIC_OUTBOX, groupId = "s13-split")
        public void onMessage(OrderCreated e,
                              @Header(HDR_MESSAGE_ID) String messageId,
                              @Header(KafkaHeaders.OFFSET) long offset) {
            // ❌ 트랜잭션 #1 — 여기서 커밋됩니다
            if (!markInSeparateTx(messageId)) {
                log.warn("duplicate messageId={} group=s13-split — skipped", messageId);
                return;
            }
            log.info("marked {} (tx#1 committed)", messageId);

            if (offset == CRASH_AT_OFFSET) {
                log.error("simulated crash before deduct");
                Runtime.getRuntime().halt(1);            // 셧다운 훅 없이 즉사
            }

            // ❌ 트랜잭션 #2 — 여기 도달 못 하면 이력만 남습니다
            deductInSeparateTx(e);
        }

        @Transactional(propagation = org.springframework.transaction.annotation.Propagation.REQUIRES_NEW)
        public boolean markInSeparateTx(String messageId) {
            try {
                jdbc.update(Sql.INSERT_MARK, markKey("s13-split", messageId), "s13-split");
                return true;
            } catch (DuplicateKeyException dup) {
                return false;
            }
        }

        @Transactional(propagation = org.springframework.transaction.annotation.Propagation.REQUIRES_NEW)
        public void deductInSeparateTx(OrderCreated e) {
            jdbc.update(Sql.DEDUCT, e.quantity(), e.sku(), e.quantity());
            log.info("deducted {} -{}", e.sku(), e.quantity());
        }
    }

    /* =====================================================================
     * [13-4] 처리 이력 테이블 청소
     * ===================================================================== */

    @Component
    @Profile({"step13-idem", "step13-order"})
    public static class ProcessedMessagePurger {

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

        /** 보관 기간 ≥ 토픽 retention(7일). 재발행·DLT 재처리 여유까지 감안해 2배로 잡습니다. */
        private static final int RETENTION_DAYS = 14;

        private final JdbcTemplate jdbc;

        public ProcessedMessagePurger(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @Scheduled(cron = "0 10 3 * * *")
        public void purge() {
            Timestamp cutoff = Timestamp.from(
                    Instant.now().minus(RETENTION_DAYS, ChronoUnit.DAYS));
            int total = 0;
            int deleted;
            do {
                // LIMIT 로 쪼갭니다. 한 번에 수천만 행을 지우면 undo 로그가 폭발하고
                // 복제 지연이 생깁니다.
                deleted = jdbc.update(Sql.PURGE_MARKS, cutoff);
                total += deleted;
            } while (deleted == 5000);
            log.info("purged {} rows from processed_message", total);
        }
    }

    /* =====================================================================
     * [13-5] Transactional Outbox
     * ===================================================================== */

    /**
     * ★ 이 클래스에 KafkaTemplate 이 주입되지 않은 것이 패턴의 전부입니다.
     * Kafka 가 죽어 있어도 주문 생성은 성공합니다.
     */
    @Component
    public static class OrderService {

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

        private final JdbcTemplate jdbc;
        private final ObjectMapper mapper;

        public OrderService(JdbcTemplate jdbc, ObjectMapper mapper) {
            this.jdbc = jdbc;
            this.mapper = mapper;
        }

        @Transactional
        public String createOrder(OrderCreated e) {
            jdbc.update(Sql.INSERT_ORDER, e.orderId(), e.customerId(), e.amount());
            jdbc.update(Sql.INSERT_OUTBOX, e.orderId(), "OrderCreated", toJson(e));
            log.info("order {} created (outbox staged)", e.orderId());
            return e.orderId();
        }

        private String toJson(OrderCreated e) {
            try {
                return mapper.writeValueAsString(e);
            } catch (Exception ex) {
                throw new IllegalStateException("cannot serialize " + e.orderId(), ex);
            }
        }
    }

    /** 릴레이를 켜고 끄는 플래그. 검증 시나리오 ④ 에서 씁니다. */
    @Component
    public static class RelayAdmin {

        private static final Logger log = LoggerFactory.getLogger(RelayAdmin.class);
        private final AtomicBoolean running = new AtomicBoolean(true);

        public boolean isRunning() {
            return running.get();
        }

        public void stop() {
            running.set(false);
            log.info("relay STOPPED");
        }

        public void start() {
            running.set(true);
            log.info("relay STARTED");
        }
    }

    @Configuration
    @EnableScheduling
    @Profile({"step13-outbox", "step13-order"})
    public static class OutboxRelay {

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

        /**
         * ★ 실험용. 이 id 를 발행한 직후, published_at 갱신 전에 프로세스를 죽입니다.
         * 재시작하면 그 이벤트가 다시 발행됩니다 = Outbox 는 at-least-once.
         * 평소에는 -1 로 두세요.
         */
        private static final long CRASH_AT_ID = -1L;

        private final JdbcTemplate jdbc;
        private final ObjectMapper mapper;
        private final KafkaTemplate<String, Object> kafkaTemplate;
        private final RelayAdmin admin;

        public OutboxRelay(JdbcTemplate jdbc, ObjectMapper mapper,
                           KafkaTemplate<String, Object> kafkaTemplate, RelayAdmin admin) {
            this.jdbc = jdbc;
            this.mapper = mapper;
            this.kafkaTemplate = kafkaTemplate;
            this.admin = admin;
        }

        @Scheduled(fixedDelay = 500)
        @Transactional                       // SELECT ... FOR UPDATE 부터 UPDATE 까지가 한 트랜잭션
        public void relay() {
            if (!admin.isRunning()) {
                return;
            }
            List<OutboxRow> rows = jdbc.query(Sql.PICK_OUTBOX, OUTBOX_ROW_MAPPER);
            if (rows.isEmpty()) {
                return;
            }

            for (OutboxRow row : rows) {
                OrderCreated event = fromJson(row.payload());

                ProducerRecord<String, Object> rec =
                        new ProducerRecord<>(TOPIC_OUTBOX, null, row.aggregateId(), event);
                // 메시지 ID = outbox 행 PK. 재발행해도 값이 안 바뀌므로 멱등 키로 적합합니다.
                rec.headers().add(HDR_MESSAGE_ID,
                        ("OBX-" + row.id()).getBytes(StandardCharsets.UTF_8));
                rec.headers().add(HDR_EVENT_TYPE,
                        row.eventType().getBytes(StandardCharsets.UTF_8));

                // ★ join() 필수. send 는 예외 없이 리턴하고 실패는 Future 안에만 있습니다(Step 02).
                //   확인 없이 published_at 을 갱신하면 발행 안 된 이벤트가 발행됨으로 표시됩니다.
                kafkaTemplate.send(rec).join();

                if (row.id() == CRASH_AT_ID) {
                    log.error("simulated crash after send, before published_at update (id={})", row.id());
                    Runtime.getRuntime().halt(1);
                }

                jdbc.update(Sql.MARK_PUBLISHED, row.id());
            }
            log.info("relayed {} events (id {}..{})",
                    rows.size(), rows.get(0).id(), rows.get(rows.size() - 1).id());
        }

        /**
         * Kafka 랙으로는 릴레이 장애가 안 보입니다. 이 게이지가 유일한 신호입니다(13-8).
         */
        @Scheduled(fixedDelay = 5000)
        public void reportBacklog() {
            Integer backlog = jdbc.queryForObject(Sql.OUTBOX_BACKLOG, Integer.class);
            if (backlog != null && backlog > 0) {
                log.info("outbox.backlog={}", backlog);
            }
        }

        private OrderCreated fromJson(String payload) {
            try {
                return mapper.readValue(payload, OrderCreated.class);
            } catch (Exception ex) {
                throw new IllegalStateException("cannot deserialize outbox payload", ex);
            }
        }
    }

    /* =====================================================================
     * [13-6] 순서 보장 관측
     * ===================================================================== */

    @Component
    @Profile("step13-outbox")
    public static class OrderingProbe {

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

        @KafkaListener(topics = TOPIC_OUTBOX, groupId = "s13-ordering", concurrency = "1")
        public void onMessage(ConsumerRecord<String, OrderCreated> rec) {
            // 같은 키는 항상 같은 파티션이므로, 여기 도착 순서가 곧 발행 순서여야 합니다.
            log.info("received {} createdAt={} partition={} offset={}",
                    rec.key(), rec.value().createdAt(), rec.partition(), rec.offset());
        }
    }

    /* =====================================================================
     * [13-7] ⚠️ 함정 — @Async / 스레드 풀은 순서와 커밋을 동시에 깨뜨린다
     * ===================================================================== */

    @Component
    @Profile("step13-async")
    public static class AsyncBadListener {

        private static final Logger log = LoggerFactory.getLogger(AsyncBadListener.class);
        private final ExecutorService pool = Executors.newFixedThreadPool(4);

        // ❌ 절대 하면 안 되는 코드
        @KafkaListener(topics = TOPIC_OUTBOX, groupId = "s13-async-bad")
        public void onMessage(ConsumerRecord<String, OrderCreated> rec) {
            log.info("submitted {} (offset={})", rec.key(), rec.offset());
            pool.submit(() -> {
                try {
                    Thread.sleep((rec.offset() % 3) * 40L);   // 스케줄링 편차를 크게
                    if (rec.offset() % 7 == 2) {
                        throw new RuntimeException("downstream timeout");
                    }
                    log.info("processed {}", rec.key());
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                } catch (RuntimeException re) {
                    // ★ 이 예외는 리스너 밖입니다. DefaultErrorHandler 가 절대 못 봅니다.
                    //   재시도도 DLT 도 안 일어나고, 오프셋은 이미 커밋됐습니다.
                    log.error("failed {}", rec.key(), re);
                }
            });
            // 여기서 즉시 리턴 → Spring 은 "처리 성공"으로 보고 커밋합니다
        }
    }

    /** ✅ 올바른 병렬화. 키 단위 레인 + 전원 완료 대기. */
    @Configuration
    @Profile("step13-async")
    public static class LaneWorkerConfig {

        private static final Logger log = LoggerFactory.getLogger(LaneWorkerConfig.class);
        private static final int WORKERS = 4;

        private final ExecutorService pool = Executors.newFixedThreadPool(WORKERS);

        @KafkaListener(topics = TOPIC_OUTBOX, groupId = "s13-async-good",
                containerFactory = "batchAckFactory")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records, Acknowledgment ack) {
            long t0 = System.currentTimeMillis();

            // 같은 키는 항상 같은 레인 → 키 안에서는 순서가 유지됩니다
            Map<Integer, List<ConsumerRecord<String, OrderCreated>>> lanes = records.stream()
                    .collect(Collectors.groupingBy(r -> Math.abs(r.key().hashCode()) % WORKERS));

            List<CompletableFuture<Void>> futures = lanes.values().stream()
                    .map(lane -> CompletableFuture.runAsync(() -> lane.forEach(this::handleOne), pool))
                    .toList();

            // ★ 이 join() 이 없으면 위의 AsyncBadListener 와 완전히 똑같아집니다
            CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();

            ack.acknowledge();
            log.info("batch={} lanes={} elapsed={}ms",
                    records.size(), lanes.size(), System.currentTimeMillis() - t0);
        }

        private void handleOne(ConsumerRecord<String, OrderCreated> rec) {
            try {
                Thread.sleep(10);                      // 처리 시간 흉내
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchAckFactory(
                ConsumerFactory<String, OrderCreated> cf) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(cf);
            f.setBatchListener(true);
            f.setConcurrency(3);
            // 레인 하나가 느리면 배치 전체가 대기합니다. max.poll.interval.ms 를 넘기지 않도록
            // 배치 크기를 줄여 둡니다.
            f.getContainerProperties().setAckMode(
                    org.springframework.kafka.listener.ContainerProperties.AckMode.MANUAL);
            return f;
        }
    }

    /* =====================================================================
     * [13-8] 최종 프로젝트 — 주문 서비스 이벤트 연동
     * ===================================================================== */

    @RestController
    @Profile("step13-order")
    public static class OrderController {

        private final OrderService orderService;
        private final RelayAdmin relayAdmin;

        public OrderController(OrderService orderService, RelayAdmin relayAdmin) {
            this.orderService = orderService;
            this.relayAdmin = relayAdmin;
        }

        // Kafka 가 죽어 있어도 200 을 반환해야 합니다. 그게 Outbox 를 쓰는 이유입니다.
        @PostMapping("/orders/{seq}")
        public String create(@PathVariable int seq) {
            return orderService.createOrder(OrderCreated.of(seq));
        }

        @PostMapping("/admin/relay/stop")
        public String stopRelay() {
            relayAdmin.stop();
            return "stopped";
        }

        @PostMapping("/admin/relay/start")
        public String startRelay() {
            relayAdmin.start();
            return "started";
        }
    }

    @Configuration
    @Profile("step13-order")
    public static class FinalWiring {

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

        private final InventoryTx tx;

        public FinalWiring(InventoryTx tx) {
            this.tx = tx;
        }

        // 체크리스트 7 — 멱등 재고 컨슈머
        @KafkaListener(topics = TOPIC_OUTBOX, groupId = GROUP_INVENTORY)
        public void inventory(OrderCreated e, @Header(HDR_MESSAGE_ID) String messageId) {
            tx.consumeOnce(messageId, GROUP_INVENTORY, e);
        }

        // 체크리스트 9 — 팬아웃. groupId 가 다르므로 같은 메시지를 각자 받습니다.
        @KafkaListener(topics = TOPIC_OUTBOX, groupId = GROUP_NOTIFICATION)
        public void notifyCustomer(OrderCreated e, @Header(HDR_MESSAGE_ID) String messageId) {
            // 알림도 멱등해야 합니다. 그룹명이 다르므로 markKey 가 달라져 서로 간섭하지 않습니다.
            log.info("notify customer={} order={} messageId={}",
                    e.customerId(), e.orderId(), messageId);
        }

        // 체크리스트 8 — 3회 재시도 후 DLT
        @Bean
        public DefaultErrorHandler finalErrorHandler(KafkaTemplate<String, Object> template) {
            DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
            return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
        }

        // DLT 관측. 재처리 도구는 연습문제 6번입니다.
        @KafkaListener(topics = TOPIC_OUTBOX_DLT, groupId = "s13-dlt-monitor")
        public void dlt(ConsumerRecord<String, OrderCreated> rec,
                        @Header(name = HDR_MESSAGE_ID, required = false) String messageId) {
            log.error("DLT messageId={} key={} partition={} offset={}",
                    messageId, rec.key(), rec.partition(), rec.offset());
        }
    }

    /** 최종 프로젝트를 CLI 없이 한 번에 돌려 보고 싶을 때. */
    @Component
    @Profile("step13-order")
    public static class SmokeRunner implements ApplicationRunner {

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

        private final OrderService orderService;

        @Value("${app.smoke.orders:0}")
        private int howMany;

        public SmokeRunner(OrderService orderService) {
            this.orderService = orderService;
        }

        @Override
        public void run(ApplicationArguments args) {
            if (howMany <= 0) {
                log.info("smoke disabled. use: --app.smoke.orders=10");
                return;
            }
            for (int i = 1; i <= howMany; i++) {
                orderService.createOrder(OrderCreated.of(i));
            }
            log.info("smoke: created {} orders", howMany);
        }
    }
}

package com.example.order.step13;

/*
 * ============================================================================
 * Step 13 — 실전 패턴과 최종 프로젝트 : Exercise (6문제)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step13/Exercise.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q1'
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q2'
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q3'
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q4'
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q4 --server.port=8081'   ← 두 번째 인스턴스
 *   ./gradlew bootRun --args='--spring.profiles.active=step13,ex13-q6'
 *   (문제 5 는 코드를 실행하지 않습니다. 주석에 답을 적는 문제입니다.)
 *
 * ★ 매 실행 전 DB 초기화
 *   mq "TRUNCATE processed_message; TRUNCATE outbox_event; DELETE FROM orders;
 *       UPDATE inventory SET quantity = 1000;"
 *
 * ★ 문제 3, 4 는 Practice.java 의 OutboxRelay 빈과 충돌합니다.
 *   ex13-* 프로필만 켜면 step13-outbox / step13-order 는 꺼지므로 그대로 두면 됩니다.
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public final class Exercise {

    private Exercise() {
    }

    static final String TOPIC_OUTBOX = "orders.outbox";
    static final String TOPIC_OUTBOX_DLT = "orders.outbox.DLT";
    static final String HDR_MESSAGE_ID = "messageId";

    /* =====================================================================
     * 문제 1. 멱등 컨슈머를 구현하고 "두 번 발행해도 한 번만 차감"을 검증하세요.
     *
     * 요구사항
     *   - processed_message 에 INSERT 를 먼저 시도하고, DuplicateKeyException 이면 스킵
     *   - 이력 INSERT 와 재고 UPDATE 는 반드시 같은 트랜잭션
     *   - 저장 키는 컨슈머 그룹을 구분할 수 있어야 함 (13-2 함정)
     *   - Publisher 가 같은 이벤트 10건을 두 번(총 20 메시지) 발행합니다.
     *     messageId 헤더는 orderId 기준으로 만들어 두 번 다 같은 값이 되게 하세요.
     * ===================================================================== */
    @Component
    @Profile("ex13-q1")
    public static class Q1IdempotentInventory {

        private static final Logger log = LoggerFactory.getLogger(Q1IdempotentInventory.class);
        static final String GROUP = "s13-ex-inventory";

        private final JdbcTemplate jdbc;

        public Q1IdempotentInventory(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @KafkaListener(topics = TOPIC_OUTBOX, groupId = GROUP)
        public void onMessage(OrderCreated e,
                              @org.springframework.messaging.handler.annotation.Header(HDR_MESSAGE_ID)
                              String messageId) {
            consumeOnce(messageId, e);
        }

        // 여기에 작성: @Transactional 을 붙이고, INSERT-first 멱등 처리 + 재고 차감을 구현하세요.
        //             (자기 호출이라 프록시를 안 타는 문제는 이 연습에서는 무시합니다.
        //              실무에서는 Practice 의 InventoryTx 처럼 별도 빈으로 빼세요.)
        public void consumeOnce(String messageId, OrderCreated e) {

        }

        // 확인:
        //   mq "SELECT sku, quantity FROM inventory ORDER BY sku;"
        //   → SKU-001=989, SKU-002=989, SKU-003=992  (20 메시지가 흘렀지만 30개만 차감)
        //   로그에 "duplicate messageId=... — skipped" 가 정확히 10줄
    }

    @Component
    @Profile("ex13-q1")
    public static class Q1Publisher implements ApplicationRunner {

        private final KafkaTemplate<String, Object> template;

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

        @Override
        public void run(ApplicationArguments args) {
            for (int round = 0; round < 2; round++) {          // ★ 같은 것을 두 번 발행
                for (int i = 1; i <= 10; i++) {
                    OrderCreated e = OrderCreated.of(i);
                    ProducerRecord<String, Object> rec =
                            new ProducerRecord<>(TOPIC_OUTBOX, null, e.orderId(), e);
                    rec.headers().add(HDR_MESSAGE_ID,
                            ("EVT-" + e.orderId()).getBytes(StandardCharsets.UTF_8));
                    template.send(rec).join();
                }
            }
        }
    }

    /* =====================================================================
     * 문제 2. SELECT-then-INSERT 방식의 경합을 재현하고 중복 차감 횟수를 세세요.
     *
     * 요구사항
     *   - 라운드마다 재고를 1000 으로 되돌리고, 두 스레드가 같은 messageId 를
     *     동시에 처리하도록 CountDownLatch 로 출발선을 맞춥니다
     *   - wrongDeduct() 는 COUNT(*) 로 먼저 확인한 뒤 차감하고 INSERT 하는 방식입니다
     *   - 100 라운드를 돌고, 재고가 997 미만인 라운드 수를 duplicateDeductions 에 셉니다
     *
     * 힌트: 카운터를 올리는 위치는 wrongDeduct() 안이 아닙니다.
     *       "INSERT 가 실패했는데 그 전에 이미 차감이 일어난" 상태는
     *       라운드가 끝난 뒤 재고를 조회해야만 판정할 수 있습니다.
     * ===================================================================== */
    @Component
    @Profile("ex13-q2")
    public static class Q2RaceCounter implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(Q2RaceCounter.class);
        private static final int ROUNDS = 100;

        private final JdbcTemplate jdbc;
        private final AtomicInteger duplicateDeductions = new AtomicInteger();

        public Q2RaceCounter(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @Override
        public void run(ApplicationArguments args) throws Exception {
            ExecutorService pool = Executors.newFixedThreadPool(2);

            for (int round = 1; round <= ROUNDS; round++) {
                String messageId = "EX2-" + round;
                jdbc.update("UPDATE inventory SET quantity = 1000 WHERE sku = 'SKU-002'");

                CountDownLatch start = new CountDownLatch(1);
                CountDownLatch done = new CountDownLatch(2);

                // 여기에 작성: 두 스레드를 pool 에 제출하고, start.await() 로 출발선을 맞춘 뒤
                //             wrongDeduct(messageId) 를 호출하게 하세요.

                start.countDown();
                done.await();

                // 여기에 작성: 재고를 조회해 997 미만이면 duplicateDeductions 를 올리세요.
            }
            pool.shutdown();
            log.warn("duplicate deductions: {} / {}", duplicateDeductions.get(), ROUNDS);
        }

        /** ❌ 틀린 방식. 손대지 마세요. 이것을 재현하는 것이 문제입니다. */
        private void wrongDeduct(String messageId) {
            Integer cnt = jdbc.queryForObject(
                    "SELECT COUNT(*) FROM processed_message WHERE message_id = ?",
                    Integer.class, messageId);
            if (cnt != null && cnt > 0) {
                return;
            }
            jdbc.update("UPDATE inventory SET quantity = quantity - 3 WHERE sku = 'SKU-002'");
            try {
                jdbc.update("INSERT INTO processed_message (message_id, consumer_group) VALUES (?, 'ex2')",
                        messageId);
            } catch (DuplicateKeyException ignored) {
                // 이미 늦었습니다
            }
        }

        // 확인:
        //   WARN ... duplicate deductions: 11 / 100     ← 숫자는 실행마다 달라집니다
        //   그 사실 자체가 답의 일부입니다. "한 번 통과했으니 괜찮다"가 왜 안 되는가?
    }

    /* =====================================================================
     * 문제 3. Transactional Outbox 를 구현하세요.
     *
     * 요구사항
     *   - Q3OrderService.createOrder() 는 @Transactional 안에서
     *     orders INSERT + outbox_event INSERT 만 합니다. Kafka 호출 금지.
     *   - Q3Relay.relay() 는 @Scheduled(fixedDelay = 500) 로
     *     published_at IS NULL 인 행을 ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED 로 집고,
     *     send().join() 후 published_at 을 갱신합니다
     *   - 발행 시 key = aggregate_id, 헤더 messageId = "OBX-" + id
     *
     * 검증
     *   docker compose stop kafka
     *   curl -s -XPOST localhost:8080/orders/1     ← 200 이 나와야 합니다
     *   mq "SELECT id, published_at FROM outbox_event;"   ← published_at IS NULL 로 쌓임
     *   docker compose start kafka                 ← 릴레이가 알아서 밀린 것을 발행
     * ===================================================================== */
    @Component
    @Profile("ex13-q3")
    public static class Q3OrderService {

        private final JdbcTemplate jdbc;
        private final ObjectMapper mapper;

        public Q3OrderService(JdbcTemplate jdbc, ObjectMapper mapper) {
            this.jdbc = jdbc;
            this.mapper = mapper;
        }

        // 여기에 작성: @Transactional 을 붙이고 orders + outbox_event 를 함께 INSERT 하세요.
        public String createOrder(OrderCreated e) {
            return e.orderId();
        }
    }

    @Configuration
    @EnableScheduling
    @Profile("ex13-q3")
    public static class Q3Relay {

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

        private final JdbcTemplate jdbc;
        private final ObjectMapper mapper;
        private final KafkaTemplate<String, Object> template;

        public Q3Relay(JdbcTemplate jdbc, ObjectMapper mapper, KafkaTemplate<String, Object> template) {
            this.jdbc = jdbc;
            this.mapper = mapper;
            this.template = template;
        }

        @Scheduled(fixedDelay = 500)
        @Transactional
        public void relay() {
            // 여기에 작성: SKIP LOCKED 로 집고 → send().join() → published_at 갱신
        }

        // 확인:
        //   INFO ... c.e.o.s13.Exercise$Q3Relay : relayed 1 events (id 1..1)
        //   mq "SELECT id, created_at, published_at FROM outbox_event;"  ← 두 시각 차이 300ms 내외
    }

    /* =====================================================================
     * 문제 4. SKIP LOCKED 를 뺀 릴레이를 두 인스턴스로 띄워 중복 발행을 측정하세요.
     *
     * 요구사항
     *   - Q4Relay 는 문제 3 과 같지만 SELECT 에서 FOR UPDATE SKIP LOCKED 를 뺍니다
     *   - 시작 시 outbox_event 에 300건을 미리 넣습니다 (seedIfEmpty)
     *   - 두 번째 인스턴스는 --server.port=8081 로 띄웁니다
     *
     * 측정
     *   kcc --topic orders.outbox --from-beginning | wc -l
     *   → 300 이어야 정상. 몇 건이 나오나요?
     *   그다음 SKIP LOCKED 를 넣고 토픽을 비운 뒤 다시 측정해 비교하세요.
     * ===================================================================== */
    @Configuration
    @EnableScheduling
    @Profile("ex13-q4")
    public static class Q4Relay {

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

        private final JdbcTemplate jdbc;
        private final ObjectMapper mapper;
        private final KafkaTemplate<String, Object> template;

        public Q4Relay(JdbcTemplate jdbc, ObjectMapper mapper, KafkaTemplate<String, Object> template) {
            this.jdbc = jdbc;
            this.mapper = mapper;
            this.template = template;
        }

        @Scheduled(fixedDelay = 200)
        @Transactional
        public void relay() {
            // 여기에 작성: FOR UPDATE SKIP LOCKED **없이** 폴링하는 릴레이
        }

        // 확인 (측정 기록):
        //   SKIP LOCKED 없음  → 토픽 메시지 수: ______   (중복 ______건)
        //   FOR UPDATE 만     → 토픽 메시지 수: ______   (락 대기로 처리량 ______)
        //   SKIP LOCKED 있음  → 토픽 메시지 수: ______
    }

    /* =====================================================================
     * 문제 5. 아래 다섯 조각 중 "순서 보장이 깨지는 것"을 모두 고르고 이유를 적으세요.
     *         코드를 실행하지 않습니다. 주석에 답만 적습니다.
     *
     *   (a) spring.kafka.listener.concurrency: 3  (파티션 3개, 키 = orderId)
     *
     *   (b) template.send("orders.outbox", event);            // 키 없이 발행
     *
     *   (c) enable.idempotence: false
     *       retries: 3
     *       max.in.flight.requests.per.connection: 5
     *
     *   (d) new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L))
     *
     *   (e) @KafkaListener(...)
     *       public void on(ConsumerRecord<String, OrderCreated> r) {
     *           CompletableFuture.runAsync(() -> handle(r), pool);
     *       }
     *
     * 답:
     *   깨지는 것 = (   ), (   ), (   )
     *   (a) 이유:
     *   (b) 이유:
     *   (c) 이유:
     *   (d) 이유:
     *   (e) 이유:
     * ===================================================================== */

    /* =====================================================================
     * 문제 6. DLT 재처리 도구를 만드세요.
     *
     * 요구사항
     *   - orders.outbox.DLT 를 읽어 원본 토픽(orders.outbox)으로 재발행합니다
     *   - ★ messageId 헤더를 반드시 보존하세요. 새 UUID 를 붙이면 멱등성이 깨져
     *     재고가 두 번 차감됩니다
     *   - kafka_dlt-* 로 시작하는 헤더는 제거합니다. 그대로 두면 다시 DLT 로 갔을 때
     *     헤더가 중첩돼 원본 정보를 잃습니다
     *   - autoStartup = "false" 로 두고, 필요할 때만 켭니다 (Step 10)
     *
     * 시나리오
     *   1) mq "UPDATE inventory SET quantity = 1 WHERE sku = 'SKU-001';"
     *   2) 주문을 넣어 DLT 로 보냅니다
     *   3) mq "UPDATE inventory SET quantity = 1000 WHERE sku = 'SKU-001';"
     *   4) 재처리 도구를 켜서 성공하는지 확인합니다
     * ===================================================================== */
    @Component
    @Profile("ex13-q6")
    public static class Q6DltReprocessor {

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

        private final KafkaTemplate<String, Object> template;

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

        @KafkaListener(id = "dltReprocessor", topics = TOPIC_OUTBOX_DLT,
                groupId = "s13-ex-dlt-reprocess", autoStartup = "false")
        public void reprocess(ConsumerRecord<String, OrderCreated> rec) {
            // 여기에 작성:
            //   1) 새 ProducerRecord 를 원본 토픽 + 원본 키로 만들고
            //   2) rec.headers() 를 순회하며 messageId 는 복사, kafka_dlt-* 는 버리고
            //   3) send().join() 으로 확인까지 하세요
        }

        /** 컨테이너를 켜는 스위치. KafkaListenerEndpointRegistry 로 시작합니다. */
        public void startReprocessing(
                org.springframework.kafka.config.KafkaListenerEndpointRegistry registry) {
            // 여기에 작성: registry 에서 "dltReprocessor" 컨테이너를 찾아 start() 하세요.
        }

        // 확인:
        //   INFO ... reprocessed messageId=OBX-11 key=ORD-0003 → orders.outbox
        //   INFO ... c.e.o.s13.InventoryTx : deducted SKU-001 -4 (remaining=996)
        //   재처리를 두 번 돌려도 재고가 한 번만 빠져야 합니다. messageId 를 보존했다면 그렇게 됩니다.
    }

    /** 문제 6 의 헤더 필터에 쓸 유틸. 참고용으로 제공합니다. */
    static boolean isDltHeader(Header h) {
        return h.key().startsWith("kafka_dlt-");
    }

    /** 문제 6 참고: 헤더 값을 문자열로 푸는 방법 */
    static String headerAsString(List<Header> headers, String key) {
        for (Header h : headers) {
            if (h.key().equals(key)) {
                return new String(h.value(), StandardCharsets.UTF_8);
            }
        }
        return null;
    }
}