#Step 01 — 환경 구축과 첫 메시지

@Component
@Profile("step01")
public class FirstPublisher implements ApplicationRunner {

    private final KafkaTemplate<String, OrderCreated> kafkaTemplate;

    public FirstPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public void run(ApplicationArguments args) {
        for (int seq = 1; seq <= 3; seq++) {
            OrderCreated event = OrderCreated.of(seq);
            kafkaTemplate.send("orders", event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) { log.error("발행 실패", ex); return; }
                    log.info("발행 성공 key={} -> {}-{}@{}",
                        event.orderId(),
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                });
        }
        kafkaTemplate.flush();
    }
}

@Component
@Profile("step01")
public class InventoryListener {

    @KafkaListener(
            id = "s01-inventory-listener",
            topics = "${app.topic.orders}",
            groupId = "s01-inventory",
            concurrency = "1")
    public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
        log.info("[1-3] 재고 차감 {}-{}@{} key={} sku={} qty={}",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value().sku(), record.value().quantity());
    }
}

Map<String, Class<?>> targets = new LinkedHashMap<>();
targets.put("KafkaTemplate",                 KafkaTemplate.class);
targets.put("ProducerFactory",               ProducerFactory.class);
targets.put("ConsumerFactory",               ConsumerFactory.class);
targets.put("KafkaListenerContainerFactory", KafkaListenerContainerFactory.class);
targets.put("KafkaAdmin",                    KafkaAdmin.class);
targets.put("KafkaListenerEndpointRegistry", KafkaListenerEndpointRegistry.class);
targets.put("KafkaTransactionManager",       KafkaTransactionManager.class);

targets.forEach((label, type) -> {
    String[] names = ctx.getBeanNamesForType(type);   // ← 구현체가 아니라 인터페이스로 조회
    ...
});

@ConditionalOnMissingBean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)

@Configuration
@EnableKafka
public class RedundantEnableKafkaConfig { }

log.info("[1-3] 재고 차감 {}-{}@{} key={} sku={} qty={}",
        record.topic(), record.partition(), record.offset(),
        record.key(), record.value().sku(), record.value().quantity());

kafkaTemplate.send("order", event.orderId(), event)   // ← 's' 가 빠졌다
        .whenComplete((result, ex) -> { ... });

package com.example.order.step01;

/*
 * ============================================================================
 * Step 01 — 환경 구축과 첫 메시지 : Practice
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step01/Practice.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step01'
 *
 * 함정 재현용 보조 프로필 (필요할 때만 함께 켭니다)
 *   ./gradlew bootRun --args='--spring.profiles.active=step01,step01-typo'
 *       → [1-8] 오타 토픽(order)으로 발행. 토픽이 조용히 새로 생깁니다.
 *   ./gradlew bootRun --args='--spring.profiles.active=step01,step01-enable-kafka'
 *       → [1-5] @EnableKafka 를 직접 붙였을 때 무슨 일이 생기는지 확인.
 *
 * 실행 전/후에 확인할 CLI (project/ 의 alias 를 등록해 두었다고 가정)
 *   kt  --list
 *   kt  --describe --topic orders
 *   kcg --describe --group s01-inventory
 *   kcc --topic orders --from-beginning --property print.key=true --property print.partition=true
 *
 * 오프셋을 처음부터 다시 읽고 싶다면 (앱을 먼저 종료할 것)
 *   kcg --group s01-inventory --topic orders --reset-offsets --to-earliest --execute
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerRecord;
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.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * Step 01 의 모든 예제를 담은 단일 파일입니다.
 * 각 nested static class 는 본문 절 번호와 1:1 로 대응합니다.
 */
public final class Practice {

    private Practice() {
        // 유틸리티 홀더. 인스턴스화하지 않습니다.
    }

    // ========================================================================
    // [1-2] 첫 발행 — KafkaTemplate 을 주입받아 orders 토픽으로 3건 보낸다
    // ========================================================================
    //
    // KafkaAutoConfiguration 이 만들어 준 kafkaTemplate 빈은 KafkaTemplate<?, ?> 로
    // 선언되어 있지만, 와일드카드는 어떤 구체 제네릭에도 대입 가능하므로
    // KafkaTemplate<String, OrderCreated> 로 주입받을 수 있습니다.
    // 실제 직렬화기는 application.yml 의 key-serializer / value-serializer 가 정합니다.
    //
    @Component
    @Profile("step01")
    @Order(20)
    public static class FirstPublisher implements ApplicationRunner {

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

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;
        private final String ordersTopic;

        public FirstPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate,
                              @Value("${app.topic.orders}") String ordersTopic) {
            this.kafkaTemplate = kafkaTemplate;
            this.ordersTopic = ordersTopic;
        }

        @Override
        public void run(ApplicationArguments args) {
            log.info("[1-2] {} 토픽으로 OrderCreated 3건 발행을 시작합니다", ordersTopic);

            for (int seq = 1; seq <= 3; seq++) {
                OrderCreated event = OrderCreated.of(seq);

                // 키를 orderId 로 줍니다. 같은 주문은 항상 같은 파티션으로 갑니다.
                CompletableFuture<SendResult<String, OrderCreated>> future =
                        kafkaTemplate.send(ordersTopic, event.orderId(), event);

                // send() 는 즉시 리턴합니다. 성공/실패는 이 콜백에서만 알 수 있습니다.
                // (콜백을 안 붙이면 실패해도 예외 없이 조용히 지나갑니다 — Step 02 의 주제)
                future.whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("[1-2] 발행 실패 key={}", event.orderId(), ex);
                        return;
                    }
                    log.info("[1-2] 발행 성공 key={} -> {}-{}@{}",
                            event.orderId(),
                            result.getRecordMetadata().topic(),
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                });
            }

            // 버퍼에 남은 레코드를 즉시 밀어냅니다. 실습에서 로그 순서를 안정시키려는 목적이며,
            // 운영 코드에서 매 건 flush() 하면 배치가 무력화되어 처리량이 급락합니다(Step 02).
            kafkaTemplate.flush();
            log.info("[1-2] flush 완료");
        }
    }

    // ========================================================================
    // [1-3] 첫 수신 — @KafkaListener
    // [1-7] 리스너 로그에 topic-partition@offset 을 항상 남긴다
    // ========================================================================
    //
    // concurrency 를 "1" 로 고정했습니다. application.yml 의 listener.concurrency=3 을
    // 그대로 두면 컨테이너 3개가 파티션을 하나씩 나눠 갖고 로그가 세 줄로 쪼개집니다.
    // Step 01 에서는 "partitions assigned: [orders-0, orders-1, orders-2]" 한 줄을
    // 눈에 익히는 것이 목적이므로 스레드를 하나로 둡니다.
    // concurrency 와 파티션 수의 관계는 Step 03 에서 다룹니다.
    //
    @Component
    @Profile("step01")
    public static class InventoryListener {

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

        @KafkaListener(
                id = "s01-inventory-listener",
                topics = "${app.topic.orders}",
                groupId = "s01-inventory",
                concurrency = "1")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            OrderCreated event = record.value();

            // [1-7] 위치 정보를 반드시 남깁니다. "메시지가 어디로 갔는가"를
            // 나중에 추적할 수 있는 가장 싼 방법입니다.
            log.info("[1-3] 재고 차감 {}-{}@{} key={} sku={} qty={}",
                    record.topic(), record.partition(), record.offset(),
                    record.key(), event.sku(), event.quantity());
        }
    }

    // ========================================================================
    // [1-4] 자동 설정이 만들어 준 빈들을 실제로 꺼내 본다
    // ========================================================================
    //
    // spring-kafka 의존성 한 줄이 컨텍스트에 무엇을 등록했는지 눈으로 확인합니다.
    // 등록되지 '않은' 빈(KafkaTransactionManager)도 함께 보여 주는 것이 핵심입니다.
    // 조건이 안 맞으면 자동 설정은 조용히 아무것도 만들지 않습니다.
    //
    @Component
    @Profile("step01")
    @Order(10)
    public static class AutoConfiguredBeanReport implements ApplicationRunner {

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

        private final ApplicationContext ctx;

        public AutoConfiguredBeanReport(ApplicationContext ctx) {
            this.ctx = ctx;
        }

        @Override
        public void run(ApplicationArguments args) {
            Map<String, Class<?>> targets = new LinkedHashMap<>();
            targets.put("KafkaTemplate", KafkaTemplate.class);
            targets.put("ProducerFactory", ProducerFactory.class);
            targets.put("ConsumerFactory", ConsumerFactory.class);
            targets.put("KafkaListenerContainerFactory", KafkaListenerContainerFactory.class);
            targets.put("KafkaAdmin", KafkaAdmin.class);
            targets.put("KafkaListenerEndpointRegistry", KafkaListenerEndpointRegistry.class);
            // transaction-id-prefix 를 설정하지 않았으므로 등록되지 않습니다(Step 09).
            targets.put("KafkaTransactionManager", KafkaTransactionManager.class);

            log.info("[1-4] ---- spring-kafka 자동 설정 빈 목록 ----");
            targets.forEach((label, type) -> {
                String[] names = ctx.getBeanNamesForType(type);
                if (names.length == 0) {
                    // 자동 설정은 조건이 안 맞으면 '조용히' 빈을 안 만듭니다.
                    log.info("[1-4] {} (없음 — 조건 불충족)", pad(label));
                    return;
                }
                for (String name : names) {
                    Class<?> impl = ctx.getType(name);
                    log.info("[1-4] {} {} -> {}", pad(label), name,
                            impl == null ? "(타입 미상)" : impl.getName());
                }
            });

            // @KafkaListener 애노테이션을 실제로 스캔하는 후처리기의 빈 이름입니다.
            // 이름이 고정되어 있다는 점이 [1-5] 의 핵심 근거입니다.
            for (String infra : new String[]{
                    "org.springframework.kafka.config.internalKafkaListenerAnnotationProcessor",
                    "org.springframework.kafka.config.internalKafkaListenerEndpointRegistry"}) {
                log.info("[1-4] {} 존재={}", pad("내부 인프라"), ctx.containsBean(infra) + " " + infra);
            }

            // 리스너 컨테이너가 몇 개 떴는지도 여기서 셀 수 있습니다.
            KafkaListenerEndpointRegistry registry =
                    ctx.getBean(KafkaListenerEndpointRegistry.class);
            Collection<MessageListenerContainer> containers = registry.getListenerContainers();
            log.info("[1-4] 리스너 컨테이너 {}개: {}", containers.size(),
                    containers.stream().map(MessageListenerContainer::getListenerId).toList());
        }

        /** 로그 열을 맞추기 위한 좌측 정렬 패딩. */
        private static String pad(String s) {
            return "%-32s".formatted(s);
        }
    }

    // ========================================================================
    // [1-5] @EnableKafka 는 왜 필요 없는가
    // ========================================================================
    //
    // Boot 를 쓰면 KafkaAnnotationDrivenConfiguration 이 이미 @EnableKafka 를 붙여 줍니다.
    // 직접 한 번 더 붙여도 KafkaListenerAnnotationBeanPostProcessor 는
    // 고정된 빈 이름(internalKafkaListenerAnnotationProcessor)으로 등록되므로
    // 중복 등록되지 않고 그냥 무시됩니다. 즉 "해롭지는 않지만 불필요"합니다.
    //
    // 이 프로필을 켜고 기동해 리스너가 여전히 정확히 한 번만 메시지를 받는지 확인하세요.
    //
    @Configuration
    @Profile("step01-enable-kafka")
    @EnableKafka
    public static class RedundantEnableKafkaConfig {
        // 의도적으로 비어 있습니다. 애노테이션만이 실험 대상입니다.
    }

    // ========================================================================
    // [1-8] 함정 재현 — 오타 난 토픽명은 조용히 새 토픽을 만든다
    // ========================================================================
    //
    // "orders" 대신 "order" 로 보냅니다. 브로커의 auto.create.topics.enable=true 때문에
    // 예외 없이 성공하고, 파티션 1개짜리 새 토픽 order 가 생깁니다.
    // 리스너는 orders 만 보고 있으므로 아무 일도 일어나지 않습니다.
    //
    // 실행 후 확인:
    //   kt --list                     → order 가 새로 생겼습니다
    //   kt --describe --topic order   → PartitionCount: 1 (orders 는 3)
    // 정리:
    //   kt --delete --topic order
    //
    @Component
    @Profile("step01-typo")
    @Order(30)
    public static class TypoTopicPublisher implements ApplicationRunner {

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

        private static final String TYPO_TOPIC = "order";   // ← 's' 가 빠졌습니다

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;

        public TypoTopicPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate) {
            this.kafkaTemplate = kafkaTemplate;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated event = OrderCreated.of(99);

            kafkaTemplate.send(TYPO_TOPIC, event.orderId(), event)
                    .whenComplete((result, ex) -> {
                        if (ex != null) {
                            log.error("[1-8] 발행 실패", ex);
                            return;
                        }
                        // 이 로그가 찍힌다는 사실 자체가 함정입니다.
                        // 오타를 냈는데 "성공"이라고 나옵니다.
                        log.warn("[1-8] 오타 토픽에 발행 '성공' {}-{}@{}",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                    });

            kafkaTemplate.flush();
            log.warn("[1-8] kt --list 로 '{}' 토픽이 생겼는지 확인하세요", TYPO_TOPIC);
        }
    }
}

package com.example.order.step01;

/*
 * ============================================================================
 * Step 01 — 환경 구축과 첫 메시지 : Exercise (문제지)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step01/Exercise.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step01ex'
 *
 * 문제별 보조 프로필
 *   step01ex-latest   문제 4 (auto-offset-reset=latest 로 메시지 유실 재현)
 *   step01ex-strict   문제 6 (오타 토픽 자동 생성 막기)
 *   step01ex-typo     문제 7 (오타 난 설정 프로퍼티가 무시되는 것 확인)
 *
 * 규칙
 *   - "여기에 작성:" 아래를 채우세요. 그 밖의 시그니처는 바꾸지 않습니다.
 *   - 매 문제 후 반드시 콘솔 로그와 CLI 출력을 눈으로 확인합니다.
 *   - 컨슈머 그룹 이름은 전부 s01- 접두사를 씁니다.
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.annotation.Order;
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.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

public final class Exercise {

    private Exercise() {
    }

    // ========================================================================
    // 문제 1. payments 토픽으로 발행하고, 전용 컨슈머 그룹으로 받으세요.
    //
    //  요구사항
    //   (a) 기동 시 payments 토픽으로 OrderCreated.of(101) 한 건을 발행합니다.
    //       토픽명은 하드코딩하지 말고 application.yml 의 app.topic.payments 를 씁니다.
    //   (b) 발행 결과(topic-partition@offset)를 로그로 남깁니다.
    //   (c) groupId = "s01-payment" 인 @KafkaListener 로 받고,
    //       역시 topic-partition@offset 을 남깁니다.
    //   (d) payments 는 파티션이 1개입니다. 기동 로그의 partitions assigned 가
    //       orders 때와 어떻게 다른지 확인하세요.
    // ========================================================================
    @Component
    @Profile("step01ex")
    @Order(10)
    public static class Ex1PaymentPublisher implements ApplicationRunner {

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

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;
        private final String paymentsTopic;

        public Ex1PaymentPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate,
                                   @Value("${app.topic.payments}") String paymentsTopic) {
            this.kafkaTemplate = kafkaTemplate;
            this.paymentsTopic = paymentsTopic;
        }

        @Override
        public void run(ApplicationArguments args) {
            // 여기에 작성: payments 토픽으로 OrderCreated.of(101) 발행 + 결과 로깅
            log.info("문제 1 미구현 (topic={})", paymentsTopic);
        }
    }

    @Component
    @Profile("step01ex")
    public static class Ex1PaymentListener {

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

        // 여기에 작성: @KafkaListener 를 붙여 payments 토픽을 s01-payment 그룹으로 구독
        public void onPayment(ConsumerRecord<String, OrderCreated> record) {
            log.info("문제 1 미구현 record={}", record);
        }
    }

    // ========================================================================
    // 문제 2. 같은 토픽을 두 컨슈머 그룹이 각각 전부 소비하는 것을 확인하세요.
    //
    //  요구사항
    //   (a) 아래 리스너를 groupId = "s01-notification" 으로 orders 토픽에 붙입니다.
    //   (b) Practice 의 s01-inventory 리스너와 동시에 켜고, 발행한 3건을
    //       두 그룹이 각각 3건씩 = 총 6줄 로그로 받는지 확인합니다.
    //   (c) kcg --list 로 두 그룹이 모두 보이는지, kcg --describe 로
    //       각 그룹의 CURRENT-OFFSET 이 독립적으로 관리되는지 확인합니다.
    //
    //  생각해 볼 것: 같은 그룹 이름을 쓰는 리스너를 하나 더 붙이면 결과가 어떻게 달라질까요?
    // ========================================================================
    @Component
    @Profile("step01ex")
    public static class Ex2NotificationListener {

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

        // 여기에 작성: @KafkaListener(topics = ..., groupId = "s01-notification", concurrency = "1")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            log.info("문제 2 미구현 record={}", record);
        }
    }

    // ========================================================================
    // 문제 3. 자동 설정이 만들어 준 빈을 직접 조회해 출력하세요.
    //
    //  요구사항
    //   (a) ApplicationContext.getBeanNamesForType 으로 아래 빈 이름을 출력합니다.
    //       KafkaTemplate / ProducerFactory / ConsumerFactory /
    //       KafkaListenerContainerFactory / KafkaAdmin / KafkaListenerEndpointRegistry
    //   (b) KafkaTransactionManager 는 왜 목록에 없는지 주석으로 한 줄 적으세요.
    //   (c) ProducerFactory 의 실제 구현 클래스 이름(getType)까지 함께 출력합니다.
    // ========================================================================
    @Component
    @Profile("step01ex")
    @Order(20)
    public static class Ex3BeanReport implements ApplicationRunner {

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

        private final ApplicationContext ctx;

        public Ex3BeanReport(ApplicationContext ctx) {
            this.ctx = ctx;
        }

        @Override
        public void run(ApplicationArguments args) {
            // 여기에 작성: 위 6종 + KafkaTransactionManager 를 조회해 출력
            log.info("문제 3 미구현 (빈 총 개수={})", ctx.getBeanDefinitionCount());
        }
    }

    // ========================================================================
    // 문제 4. auto-offset-reset = latest 로 "메시지 유실"을 재현하세요.
    //
    //  요구사항
    //   (a) 아래 팩토리에서 AUTO_OFFSET_RESET_CONFIG 를 "latest" 로 설정합니다.
    //       나머지 설정은 KafkaProperties(=application.yml)에서 그대로 가져옵니다.
    //   (b) 리스너에 containerFactory = "latestContainerFactory" 를 지정하고
    //       groupId 는 "s01-latest" 로 둡니다. (한 번도 커밋한 적 없는 새 그룹)
    //   (c) 실행 절차
    //        1. 앱을 끈 상태에서 콘솔 프로듀서로 orders 에 메시지를 3건 넣습니다.
    //        2. 앱을 켭니다.  → 리스너에 아무것도 안 들어옵니다.
    //        3. 앱을 켠 채로 다시 3건을 넣습니다. → 이번엔 들어옵니다.
    //   (d) 그 뒤 앱을 껐다 켜고 1번을 반복하면 이번엔 메시지가 들어옵니다.
    //       왜 그런지 주석으로 설명하세요. (힌트: 커밋된 오프셋의 유무)
    // ========================================================================
    @Configuration
    @Profile("step01ex-latest")
    public static class Ex4LatestConfig {

        @Bean
        public ConsumerFactory<String, OrderCreated> latestConsumerFactory(KafkaProperties props) {
            // Boot 3.2.x 는 인자 없는 buildConsumerProperties() 입니다.
            // (SslBundles 를 받는 오버로드는 Boot 3.4 부터입니다.)
            Map<String, Object> config = new HashMap<>(props.buildConsumerProperties());
            // 여기에 작성: ConsumerConfig.AUTO_OFFSET_RESET_CONFIG 를 "latest" 로
            config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // ← 고치세요
            return new DefaultKafkaConsumerFactory<>(config);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> latestContainerFactory(
                ConsumerFactory<String, OrderCreated> latestConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(latestConsumerFactory);
            factory.setConcurrency(1);
            return factory;
        }
    }

    @Component
    @Profile("step01ex-latest")
    public static class Ex4LatestListener {

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

        // 여기에 작성: containerFactory = "latestContainerFactory", groupId = "s01-latest"
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            log.info("문제 4 미구현 record={}", record);
        }
    }

    // ========================================================================
    // 문제 5. 리스너 로그를 "추적 가능한" 형태로 만드세요.
    //
    //  요구사항
    //   (a) topic-partition@offset, key, 그리고 레코드 타임스탬프를 한 줄에 남깁니다.
    //   (b) 포맷은 "orders-1@42 key=ORD-0001 ts=..." 처럼 사람이 grep 할 수 있게 합니다.
    //   (c) 로그 한 줄만 보고 kcc / kcg 로 같은 레코드를 찾아갈 수 있어야 합니다.
    // ========================================================================
    @Component
    @Profile("step01ex")
    public static class Ex5TraceableListener {

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

        @KafkaListener(
                id = "s01ex-trace",
                topics = "${app.topic.orders}",
                groupId = "s01-trace",
                concurrency = "1")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            // 여기에 작성: topic-partition@offset + key + timestamp 를 한 줄로
            log.info("문제 5 미구현");
        }
    }

    // ========================================================================
    // 문제 6. 오타 난 토픽으로 보냈을 때 '조용히 성공'하지 않게 만드세요.
    //
    //  요구사항
    //   (a) 존재하지 않는 토픽 "order" 로 발행을 시도합니다.
    //   (b) 기본 상태에서는 토픽이 자동 생성되며 성공합니다. 먼저 그것을 확인하세요.
    //   (c) 브로커의 auto.create.topics.enable 을 끄지 않고, 애플리케이션 쪽 설정만으로
    //       "실패하게" 만드는 방법을 찾아 적용하세요.
    //       힌트 1: 컨슈머에는 allow.auto.create.topics 프로퍼티가 있습니다.
    //       힌트 2: 리스너 컨테이너에는 missing-topics-fatal 이 있습니다.
    //       힌트 3: 프로듀서는 어떤가요? 여기서 얻는 교훈이 이 문제의 핵심입니다.
    //   (d) 확인 후 kt --delete --topic order 로 정리합니다.
    // ========================================================================
    @Component
    @Profile("step01ex-strict")
    @Order(30)
    public static class Ex6TypoTopicPublisher implements ApplicationRunner {

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

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;

        public Ex6TypoTopicPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate) {
            this.kafkaTemplate = kafkaTemplate;
        }

        @Override
        public void run(ApplicationArguments args) {
            // 여기에 작성: "order" 토픽으로 발행하고 결과를 로깅
            log.info("문제 6 미구현 template={}", kafkaTemplate.getDefaultTopic());
        }
    }

    // ========================================================================
    // 문제 7. 오타 난 설정 프로퍼티가 어떻게 처리되는지 확인하세요.
    //
    //  요구사항
    //   (a) application.yml 의 spring.kafka.producer.properties 아래에
    //       lingerms: 5   (점이 빠진 오타) 를 추가합니다.
    //   (b) 앱을 기동해 무슨 일이 생기는지 확인합니다. 예외가 납니까?
    //   (c) 기동 로그에서 그 오타를 알려 주는 단 한 줄을 찾아 아래 상수에 적으세요.
    //   (d) 그 한 줄이 WARN 레벨이라는 사실이 왜 위험한지 주석으로 적으세요.
    // ========================================================================
    @Component
    @Profile("step01ex-typo")
    @Order(40)
    public static class Ex7UnknownConfigProbe implements ApplicationRunner {

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

        // 여기에 작성: 기동 로그에서 찾은 경고 한 줄을 그대로 붙여넣으세요.
        private static final String EXPECTED_WARNING = "(여기에 작성)";

        @Override
        public void run(ApplicationArguments args) {
            log.info("문제 7 — 기동 로그에서 찾아야 할 경고: {}", EXPECTED_WARNING);
        }
    }
}

package com.example.order.step01;

/*
 * ============================================================================
 * Step 01 — 환경 구축과 첫 메시지 : Solution (정답과 해설)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step01/Solution.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step01sol'
 *   ./gradlew bootRun --args='--spring.profiles.active=step01sol,step01sol-latest'
 *   ./gradlew bootRun --args='--spring.profiles.active=step01sol,step01sol-strict'
 *
 * Exercise 를 직접 풀어 본 뒤에 여세요.
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

public final class Solution {

    private Solution() {
    }

    // ========================================================================
    // 정답 1 — payments 토픽 발행 + s01-payment 그룹 수신
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - 토픽명을 코드에 하드코딩하지 않고 @Value("${app.topic.payments}") 로 받는 이유는
     *    단순한 취향이 아닙니다. Kafka 는 토픽명 오타를 예외로 알려 주지 않습니다.
     *    브로커의 auto.create.topics.enable 이 켜져 있으면 새 토픽이 조용히 생기고,
     *    발행은 '성공'합니다. 토픽명을 한 곳(application.yml)에서만 관리하면
     *    프로듀서와 컨슈머가 같은 문자열을 보게 되어 이 사고 유형이 원천 차단됩니다.
     *
     *  - send() 는 CompletableFuture 를 리턴하고 즉시 반환됩니다. whenComplete 를
     *    붙이지 않으면 브로커가 거절해도 애플리케이션은 아무것도 모릅니다.
     *    Step 01 단계에서부터 "발행 결과를 반드시 본다"를 습관으로 만드세요.
     *
     *  - payments 는 파티션이 1개입니다. 그래서 기동 로그가
     *      s01-payment: partitions assigned: [payments-0]
     *    한 줄로 끝납니다. orders(3파티션)는 세 개가 한 줄에 나열됩니다.
     *    partitions assigned 로그는 "무엇을 몇 개 할당받았는가"를 보여 주는 창이며,
     *    여기가 비어 있으면([]) 리스너는 살아 있어도 아무것도 받지 못합니다.
     */
    @Component
    @Profile("step01sol")
    @Order(10)
    public static class Sol1PaymentPublisher implements ApplicationRunner {

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

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;
        private final String paymentsTopic;

        public Sol1PaymentPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate,
                                    @Value("${app.topic.payments}") String paymentsTopic) {
            this.kafkaTemplate = kafkaTemplate;
            this.paymentsTopic = paymentsTopic;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated event = OrderCreated.of(101);

            kafkaTemplate.send(paymentsTopic, event.orderId(), event)
                    .whenComplete((result, ex) -> {
                        if (ex != null) {
                            log.error("[정답1] 발행 실패 key={}", event.orderId(), ex);
                            return;
                        }
                        log.info("[정답1] 발행 성공 {}-{}@{} key={}",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset(),
                                event.orderId());
                    });
            kafkaTemplate.flush();
        }
    }

    @Component
    @Profile("step01sol")
    public static class Sol1PaymentListener {

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

        @KafkaListener(
                id = "s01sol-payment",
                topics = "${app.topic.payments}",
                groupId = "s01-payment",
                concurrency = "1")
        public void onPayment(ConsumerRecord<String, OrderCreated> record) {
            log.info("[정답1] 결제 수신 {}-{}@{} key={} amount={}",
                    record.topic(), record.partition(), record.offset(),
                    record.key(), record.value().amount());
        }
    }

    // ========================================================================
    // 정답 2 — 컨슈머 그룹 두 개는 같은 메시지를 각자 전부 받는다
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - groupId 만 다르게 준 리스너를 하나 더 붙이면 끝입니다. 코드에 특별한 것은 없습니다.
     *    중요한 건 결과입니다. 3건을 발행하면 로그가 6줄 나옵니다.
     *    s01-inventory 가 3건, s01-notification 이 3건. 이것이 Kafka 의 pub/sub 입니다.
     *
     *  - 오프셋은 (그룹, 토픽, 파티션) 조합마다 따로 관리됩니다. 그래서
     *      kcg --describe --group s01-inventory
     *      kcg --describe --group s01-notification
     *    의 CURRENT-OFFSET 은 서로 완전히 독립적입니다. 한쪽을 리셋해도
     *    다른 쪽은 영향을 받지 않습니다.
     *
     *  - 반대로 groupId 를 '같게' 주고 리스너를 둘 붙이면 어떻게 될까요?
     *    두 리스너가 한 그룹의 두 멤버가 되어 파티션을 나눠 갖습니다.
     *    각 메시지는 둘 중 하나에게만 갑니다. 총 로그는 6줄이 아니라 3줄입니다.
     *    "왜 절반만 처리되지?" 라는 질문의 절반은 여기서 나옵니다.
     *    파티션이 3개인데 같은 그룹 멤버가 4개면 한 명은 아무것도 못 받고 놉니다(Step 03).
     */
    @Component
    @Profile("step01sol")
    public static class Sol2NotificationListener {

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

        @KafkaListener(
                id = "s01sol-notification",
                topics = "${app.topic.orders}",
                groupId = "s01-notification",
                concurrency = "1")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            log.info("[정답2] 알림 발송 {}-{}@{} key={} customerId={}",
                    record.topic(), record.partition(), record.offset(),
                    record.key(), record.value().customerId());
        }
    }

    // ========================================================================
    // 정답 3 — 자동 설정 빈 목록
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - getBeanNamesForType 은 인터페이스로 조회해야 합니다. ProducerFactory 는
     *    인터페이스이고 실제 구현은 DefaultKafkaProducerFactory 입니다.
     *    DefaultKafkaProducerFactory.class 로 조회해도 지금은 찾아지지만,
     *    커스텀 구현으로 바꾸는 순간 조회가 깨집니다. 인터페이스로 조회하세요.
     *
     *  - KafkaTransactionManager 가 목록에 없는 이유:
     *    KafkaAutoConfiguration 의 해당 @Bean 에는
     *      @ConditionalOnProperty(name = "spring.kafka.producer.transaction-id-prefix")
     *    가 붙어 있습니다. application.yml 에 transaction-id-prefix 를 안 적었으므로
     *    조건이 불충족이고, 빈은 '조용히' 만들어지지 않습니다.
     *    이것이 Spring Boot 자동 설정의 기본 성격입니다. 실패가 아니라 부재입니다.
     *    Step 09 에서 transaction-id-prefix 를 넣는 순간 이 빈이 나타납니다.
     *
     *  - ctx.getType(name) 이 null 을 리턴할 수 있다는 점에 유의하세요.
     *    빈 정의만 있고 아직 타입이 확정되지 않은 경우입니다. 방어 코드를 넣습니다.
     */
    @Component
    @Profile("step01sol")
    @Order(20)
    public static class Sol3BeanReport implements ApplicationRunner {

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

        private final ApplicationContext ctx;

        public Sol3BeanReport(ApplicationContext ctx) {
            this.ctx = ctx;
        }

        @Override
        public void run(ApplicationArguments args) {
            Map<String, Class<?>> targets = new LinkedHashMap<>();
            targets.put("KafkaTemplate", KafkaTemplate.class);
            targets.put("ProducerFactory", ProducerFactory.class);
            targets.put("ConsumerFactory", ConsumerFactory.class);
            targets.put("KafkaListenerContainerFactory", KafkaListenerContainerFactory.class);
            targets.put("KafkaAdmin", KafkaAdmin.class);
            targets.put("KafkaListenerEndpointRegistry", KafkaListenerEndpointRegistry.class);
            targets.put("KafkaTransactionManager", KafkaTransactionManager.class);

            targets.forEach((label, type) -> {
                String[] names = ctx.getBeanNamesForType(type);
                if (names.length == 0) {
                    log.info("[정답3] {} → (없음) 조건이 충족되지 않아 등록되지 않았습니다",
                            "%-32s".formatted(label));
                    return;
                }
                for (String name : names) {
                    Class<?> impl = ctx.getType(name);
                    log.info("[정답3] {} → {} ({})", "%-32s".formatted(label), name,
                            impl == null ? "타입 미확정" : impl.getSimpleName());
                }
            });
        }
    }

    // ========================================================================
    // 정답 4 — auto-offset-reset = latest 로 메시지 유실 재현
    // ========================================================================
    /*
     * 왜 이 답인가 — 이 스텝에서 가장 중요한 정답입니다.
     *
     *  - auto.offset.reset 은 "커밋된 오프셋이 없을 때 어디서부터 읽을지"만 정합니다.
     *    항상 적용되는 값이 아닙니다. 이 한 문장을 놓치면 실습 결과가 뒤죽박죽이 됩니다.
     *
     *  - latest 로 두고 새 그룹(s01-latest)으로 처음 붙으면, 컨슈머는
     *    "지금 이 순간의 LOG-END-OFFSET" 부터 읽기 시작합니다.
     *    그러니 앱을 켜기 전에 넣어 둔 메시지는 영원히 안 보입니다.
     *    로그는 완벽하게 깨끗합니다. 예외도 없고 경고도 없습니다.
     *    partitions assigned 도 정상적으로 찍힙니다. 그래서 코드를 의심하게 됩니다.
     *
     *  - 앱을 켠 상태에서 넣은 메시지는 정상적으로 들어옵니다.
     *    그리고 그 순간 s01-latest 그룹에 오프셋이 '커밋'됩니다.
     *    이제부터는 auto.offset.reset 이 아예 개입하지 않습니다.
     *    앱을 껐다 켠 뒤 메시지를 넣으면, 커밋된 오프셋부터 이어 읽으므로
     *    latest 인데도 "앱이 꺼진 동안 들어온 메시지"가 보입니다.
     *    문제 4(d)의 답이 이것입니다. 값이 바뀐 게 아니라 조건이 사라진 것입니다.
     *
     *  - 진단 방법: kcg --describe --group s01-latest 를 실행해서
     *    그 그룹이 아예 없다고 나오면 '한 번도 커밋한 적 없는' 상태입니다.
     *    이 경우에만 auto.offset.reset 이 작동합니다.
     *
     *  - 운영 기본값이 latest 인 것은 이유가 있습니다. 새 서비스를 붙일 때
     *    earliest 면 몇 달치 과거 메시지가 한꺼번에 쏟아집니다.
     *    "학습에는 earliest, 운영에는 대개 latest, 단 반드시 의식하고 고를 것" 입니다.
     */
    @Configuration
    @Profile("step01sol-latest")
    public static class Sol4LatestConfig {

        @Bean
        public ConsumerFactory<String, OrderCreated> latestConsumerFactory(KafkaProperties props) {
            Map<String, Object> config = new HashMap<>(props.buildConsumerProperties());
            config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
            return new DefaultKafkaConsumerFactory<>(config);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> latestContainerFactory(
                ConsumerFactory<String, OrderCreated> latestConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(latestConsumerFactory);
            factory.setConcurrency(1);
            return factory;
        }
    }

    @Component
    @Profile("step01sol-latest")
    public static class Sol4LatestListener {

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

        @KafkaListener(
                id = "s01sol-latest",
                topics = "${app.topic.orders}",
                groupId = "s01-latest",
                containerFactory = "latestContainerFactory")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            log.info("[정답4] 수신 {}-{}@{} key={}",
                    record.topic(), record.partition(), record.offset(), record.key());
        }
    }

    // ========================================================================
    // 정답 5 — 추적 가능한 리스너 로그
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - "orders-1@42" 라는 좌표 하나가 있으면 그 레코드를 언제든 다시 찾아갈 수 있습니다.
     *      kcc --topic orders --partition 1 --offset 42 --max-messages 1
     *    반대로 좌표가 없으면 "그 메시지가 처리됐는지"를 사후에 증명할 방법이 없습니다.
     *
     *  - 구분자를 '-' 와 '@' 로 고정하는 것이 중요합니다. Spring Kafka 자신이
     *    에러 로그에서 이 포맷을 씁니다.
     *      Backoff ... exhausted for orders-1@42
     *    애플리케이션 로그와 프레임워크 로그의 좌표 포맷이 같으면
     *    한 번의 grep 으로 양쪽을 동시에 훑을 수 있습니다.
     *
     *  - record.timestamp() 는 기본적으로 프로듀서가 찍은 시각(CreateTime)입니다.
     *    브로커 도착 시각을 원하면 토픽 설정 message.timestamp.type 을 바꿔야 합니다.
     *    "내 로그 시각 - 레코드 타임스탬프" 가 그 레코드의 실제 지연(end-to-end lag)입니다.
     *    컨슈머 랙(건수)과 지연(시간)은 다른 지표이며, 둘 다 필요합니다(Step 12).
     */
    @Component
    @Profile("step01sol")
    public static class Sol5TraceableListener {

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

        @KafkaListener(
                id = "s01sol-trace",
                topics = "${app.topic.orders}",
                groupId = "s01-trace",
                concurrency = "1")
        public void onOrderCreated(ConsumerRecord<String, OrderCreated> record) {
            Instant produced = Instant.ofEpochMilli(record.timestamp());
            long lagMs = System.currentTimeMillis() - record.timestamp();

            log.info("[정답5] {}-{}@{} key={} ts={} lagMs={}",
                    record.topic(), record.partition(), record.offset(),
                    record.key(), produced, lagMs);
        }
    }

    // ========================================================================
    // 정답 6 — 오타 토픽을 '조용히 성공'하지 않게 만들기
    // ========================================================================
    /*
     * 왜 이 답인가 — 그리고 이 문제의 진짜 답은 "완전히 막을 수는 없다" 입니다.
     *
     *  (1) 컨슈머 쪽: allow.auto.create.topics = false
     *      컨슈머가 없는 토픽을 구독할 때 브로커에 토픽 생성을 요청하지 않게 합니다.
     *      여기에 listener.missing-topics-fatal = true 를 더하면,
     *      리스너가 구독하는 토픽이 없을 때 기동 자체가 실패합니다.
     *      "오타 난 토픽을 구독하는" 사고는 이걸로 잡힙니다.
     *      기동 로그:
     *        IllegalStateException: Topic(s) [order] is/are not present and missingTopicsFatal is true
     *
     *  (2) 프로듀서 쪽: 막을 수 있는 클라이언트 설정이 없습니다.
     *      프로듀서는 메타데이터를 조회할 뿐이고, 없는 토픽을 만들지 말지는
     *      전적으로 브로커의 auto.create.topics.enable 이 결정합니다.
     *      즉 "오타 난 토픽으로 send 하는" 사고는 애플리케이션 설정만으로는 못 막습니다.
     *
     *  (3) 그래서 실무의 정답은 셋입니다.
     *      - 운영 브로커에서는 auto.create.topics.enable = false 로 둔다 (가장 확실)
     *      - 토픽명을 문자열 리터럴로 쓰지 않는다. 설정값이나 상수 한 곳에서만 관리한다
     *      - NewTopic 빈으로 쓸 토픽을 선언해 두고, 그 목록 밖의 토픽은 코드 리뷰에서 잡는다
     *
     *  (4) 아래 코드는 (1)을 구현하고, 프로듀서 쪽은 여전히 성공한다는 사실을
     *      로그로 드러내 보여 줍니다. 이 비대칭이 이 문제의 학습 포인트입니다.
     */
    @Configuration
    @Profile("step01sol-strict")
    public static class Sol6StrictConfig {

        @Bean
        public ConsumerFactory<String, OrderCreated> strictConsumerFactory(KafkaProperties props) {
            Map<String, Object> config = new HashMap<>(props.buildConsumerProperties());
            config.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
            return new DefaultKafkaConsumerFactory<>(config);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> strictContainerFactory(
                ConsumerFactory<String, OrderCreated> strictConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(strictConsumerFactory);
            factory.setConcurrency(1);
            // 토픽이 없으면 기동을 실패시킵니다. application.yml 의
            // spring.kafka.listener.missing-topics-fatal 은 기본 컨테이너 팩토리에만
            // 적용되므로, 직접 만든 팩토리에는 이렇게 코드로 지정합니다.
            factory.getContainerProperties().setMissingTopicsFatal(true);
            return factory;
        }
    }

    @Component
    @Profile("step01sol-strict")
    @Order(30)
    public static class Sol6TypoTopicPublisher implements ApplicationRunner {

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

        private static final String TYPO_TOPIC = "order";

        private final KafkaTemplate<String, OrderCreated> kafkaTemplate;

        public Sol6TypoTopicPublisher(KafkaTemplate<String, OrderCreated> kafkaTemplate) {
            this.kafkaTemplate = kafkaTemplate;
        }

        @Override
        public void run(ApplicationArguments args) {
            OrderCreated event = OrderCreated.of(99);

            kafkaTemplate.send(TYPO_TOPIC, event.orderId(), event)
                    .whenComplete((result, ex) -> {
                        if (ex != null) {
                            log.error("[정답6] 발행 실패 — 브로커가 토픽 자동 생성을 거부했습니다", ex);
                            return;
                        }
                        log.warn("[정답6] 프로듀서는 여전히 성공합니다: {}-{}@{}",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                        log.warn("[정답6] 프로듀서 쪽 오타는 클라이언트 설정으로 못 막습니다. "
                                + "브로커의 auto.create.topics.enable=false 가 유일한 방어선입니다");
                    });
            kafkaTemplate.flush();
        }
    }

    // ========================================================================
    // 정답 7 — 오타 난 설정 프로퍼티는 WARN 한 줄로 끝난다
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - 찾아야 할 로그 한 줄은 이것입니다.
     *      WARN  o.a.k.c.p.ProducerConfig :
     *            The configuration 'lingerms' was supplied but isn't a known config.
     *
     *  - Kafka 클라이언트는 자기가 모르는 설정 키를 받으면 예외를 던지지 않습니다.
     *    AbstractConfig 가 사용되지 않은 키를 모아 WARN 으로 한 번 알리고 끝입니다.
     *    설계 의도는 호환성입니다. 신버전 설정을 구버전 클라이언트에 줘도 죽지 않아야 하고,
     *    인터셉터·시리얼라이저가 자기 설정을 자유롭게 끼워 넣을 수 있어야 하니까요.
     *
     *  - 위험한 이유는 세 가지입니다.
     *      1. WARN 이라서 대부분의 운영 로그 수집기에서 알림이 안 갑니다.
     *      2. 기동 시 딱 한 번만 나오고, 그 앞뒤가 ProducerConfig 설정 덤프라
     *         수십 줄에 파묻힙니다.
     *      3. 증상이 "안 됨"이 아니라 "기본값으로 동작함"입니다.
     *         linger.ms 를 5 로 주려다 오타를 내면 기본값 0 으로 돕니다.
     *         에러가 없으니 성능이 이상해도 설정을 의심하지 않게 됩니다.
     *
     *  - 방어책
     *      - 기동 로그에서 "isn't a known config" 를 grep 하는 것을 배포 체크리스트에 넣습니다.
     *      - 가능하면 ProducerConfig.LINGER_MS_CONFIG 같은 상수를 코드에서 씁니다.
     *        오타가 컴파일 에러가 됩니다. yml 문자열은 오타가 런타임 침묵이 됩니다.
     *      - Boot 의 축약 프로퍼티(spring.kafka.producer.acks 등)는 오타 나면
     *        Boot 가 잡아 줍니다. properties: 아래 원본 키만 무방비입니다.
     */
    @Component
    @Profile("step01sol")
    @Order(40)
    public static class Sol7UnknownConfigProbe implements ApplicationRunner {

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

        private static final String EXPECTED_WARNING =
                "The configuration 'lingerms' was supplied but isn't a known config.";

        @Override
        public void run(ApplicationArguments args) {
            log.info("[정답7] 기동 로그에서 찾아야 할 한 줄: {}", EXPECTED_WARNING);
            log.info("[정답7] 확인 명령: ./gradlew bootRun ... 2>&1 | grep \"isn't a known config\"");
        }
    }
}