#Step 06 — 오프셋 커밋과 AckMode

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> recordAckFactory() {
    var f = new ConcurrentKafkaListenerContainerFactory<String, OrderCreated>();
    f.setConsumerFactory(consumerFactory);
    f.setConcurrency(1);
    f.getContainerProperties().setAckMode(AckMode.RECORD);
    return f;
}

@KafkaListener(topics = "orders", groupId = "s06-record", containerFactory = "recordAckFactory")
public void onMessage(OrderCreated event) { ... }

public void onMessage(ConsumerRecord<String, OrderCreated> record) {   // AckMode.BATCH
    log.info("[6-5] 처리 {} (누적 호출 {}회) partition={} offset={}", ...);
    if (seq == 5) {
        throw new IllegalStateException("재고 서비스 응답 없음: " + record.value().orderId());
    }
}

public void onMessage(ConsumerRecord<String, OrderCreated> record) {   // AckMode.BATCH
    try {
        handle(record.value());
    } catch (RuntimeException ex) {
        // ⚠️ 이 한 줄이 메시지를 지웁니다.
        log.error("[6-5] 처리 실패, 일단 넘어갑니다: {}", record.value().orderId(), ex);
    }
}

ack.nack(Duration.ofSeconds(1));          // 레코드 리스너용: 현재 레코드부터 재처리, 1초 대기
ack.nack(47, Duration.ofSeconds(1));      // 배치 리스너용: index 47 부터 재처리, 0~46 은 커밋

@KafkaListener(id = "s06-forget", ..., containerFactory = "manualAckFactory")
public void onMessage(OrderCreated event, Acknowledgment ack) {
    log.debug("[6-7] 처리 {}", event.orderId());
    // ⚠️ ack.acknowledge(); 를 부르지 않았습니다.
}

for (int i = 0; i < records.size(); i++) {
    if (i == 47) {
        ack.nack(i, Duration.ofSeconds(1));   // 0~46 커밋, 47 부터 재시도
        return;                               // ⚠️ 반드시 즉시 리턴
    }
    totalHandled.incrementAndGet();
}
ack.acknowledge();                            // 끝까지 갔으면 배치 전체 커밋

throw new BatchListenerFailedException("결제 승인 거절: " + record.value().orderId(), i);

package com.example.order.step06;

/*
 * ============================================================================
 * Step 06 — 오프셋 커밋과 AckMode : Practice
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step06/Practice.java
 *
 * 기본 실행 (시드 데이터만 넣고 아무 리스너도 켜지 않습니다)
 *   ./gradlew bootRun --args='--spring.profiles.active=step06'
 *
 * 절별 보조 프로필 — 반드시 "하나씩" 켤 것. 동시에 켜면 커밋 로그가 뒤섞입니다.
 *   step06,step06-record            → [6-3][6-4] AckMode.RECORD           (커밋 500회)
 *   step06,step06-batch             → [6-3][6-4] AckMode.BATCH (기본값)   (커밋 1회)
 *   step06,step06-time              → [6-3]      AckMode.TIME  (5초마다)
 *   step06,step06-count             → [6-3]      AckMode.COUNT (100건마다)
 *   step06,step06-swallow           → [6-5] ⚠️ 예외를 삼켜 메시지를 "유실"시킨다
 *   step06,step06-rethrow           → [6-5]    예외를 던져 앞부분이 "중복" 처리된다
 *   step06,step06-manual            → [6-6] MANUAL_IMMEDIATE + 조건부 커밋
 *   step06,step06-forget            → [6-7] ⚠️ acknowledge() 를 안 부른다 → LAG 이 안 준다
 *   step06,step06-batch-whole       → [6-8] ⚠️ 배치 부분 실패 → 100건 통째로 재처리
 *   step06,step06-batch-nack        → [6-8] ① ack.nack(index, Duration) 로 부분 커밋
 *   step06,step06-batch-failedex    → [6-8] ② BatchListenerFailedException 으로 부분 커밋
 *
 * 커밋 시점을 눈으로 보려면 application.yml 에 아래를 추가하세요 ([6-4]).
 *   logging:
 *     level:
 *       org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: DEBUG
 *
 * 실행 전/후에 확인할 CLI (project/ 의 alias 를 등록해 두었다고 가정)
 *   kcg --list
 *   kcg --describe --group s06-inventory
 *   kcg --describe --group s06-manual
 *   kt  --describe --topic orders
 *
 * 오프셋을 처음부터 다시 읽고 싶다면 (앱을 먼저 종료할 것)
 *   kcg --group s06-inventory --topic orders --reset-offsets --to-earliest --execute
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
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.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.KafkaTemplate;
import org.springframework.kafka.listener.BatchListenerFailedException;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Step 06 의 모든 예제를 담은 단일 파일입니다.
 * 각 nested static class 는 본문 절 번호와 1:1 로 대응합니다.
 *
 * 이 스텝의 리스너는 전부 concurrency=1 로 고정된 전용 팩터리를 씁니다.
 * 커밋 로그를 읽는 것이 목적인데, 스레드가 3개면 세 컨슈머의 커밋이 섞여
 * "몇 번 커밋했는가"를 셀 수 없기 때문입니다.
 */
public final class Practice {

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

    static final String TOPIC = "orders";

    // ========================================================================
    // [6-0] 실습 준비 — orders 토픽에 결정적 시드 데이터 500건을 넣는다
    // ========================================================================
    //
    // OrderCreated.of(seq) 는 난수를 쓰지 않으므로 몇 번을 실행하든 같은 값이고,
    // 키가 같으니 파티션 배정도 항상 같습니다.
    // 이미 500건이 들어 있다면 이 러너를 건너뛰도록 --app.step06.seed=false 를 주세요.
    //
    @Component
    @Profile("step06")
    @Order(1)
    public static class Seeder implements ApplicationRunner {

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

        private final KafkaTemplate<String, OrderCreated> template;

        public Seeder(KafkaTemplate<String, OrderCreated> template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) {
            if (args.containsOption("app.step06.seed")
                    && "false".equals(args.getOptionValues("app.step06.seed").get(0))) {
                log.info("[6-0] 시드 생략 (--app.step06.seed=false)");
                return;
            }
            long t0 = System.nanoTime();
            for (int seq = 1; seq <= 500; seq++) {
                OrderCreated event = OrderCreated.of(seq);
                template.send(TOPIC, event.orderId(), event);
            }
            template.flush();
            long ms = (System.nanoTime() - t0) / 1_000_000;
            log.info("[6-0] 시드 완료: 500건 발행, {} ms", ms);
        }
    }

    // ========================================================================
    // [6-3] AckMode 6종을 각각 별도 컨테이너 팩터리로 만든다
    // ========================================================================
    //
    // application.yml 의 spring.kafka.listener.ack-mode 는 "기본 팩터리"에만 적용됩니다.
    // 한 애플리케이션 안에서 모드를 비교하려면 팩터리를 따로 만들고
    // @KafkaListener(containerFactory = "...") 로 지목하는 것이 가장 확실합니다.
    //
    // 주의: AckMode 는 Kafka 클라이언트가 아니라 "Spring 리스너 컨테이너"의 개념입니다.
    //       브로커에 전달되는 설정이 아니라, 컨테이너가 commitSync/commitAsync 를
    //       "언제 호출할지"를 정하는 정책입니다.
    //
    @Configuration
    @Profile("step06")
    public static class AckModeFactories {

        private final ConsumerFactory<String, OrderCreated> consumerFactory;

        public AckModeFactories(ConsumerFactory<String, OrderCreated> consumerFactory) {
            this.consumerFactory = consumerFactory;
        }

        private ConcurrentKafkaListenerContainerFactory<String, OrderCreated> base(AckMode mode) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            f.setConcurrency(1);                       // 커밋 횟수를 세려면 스레드 1개여야 합니다
            f.getContainerProperties().setAckMode(mode);
            return f;
        }

        /** 레코드 1건 처리할 때마다 커밋. 가장 안전하고 가장 느립니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> recordAckFactory() {
            return base(AckMode.RECORD);
        }

        /** poll 로 가져온 배치를 다 처리하면 한 번 커밋. spring-kafka 의 기본값입니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchAckFactory() {
            return base(AckMode.BATCH);
        }

        /** ackTime(ms) 이 지난 뒤 오는 첫 커밋 기회에 커밋. "5초마다"가 아니라 "5초 지나면"입니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> timeAckFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = base(AckMode.TIME);
            f.getContainerProperties().setAckTime(5_000L);
            return f;
        }

        /** ackCount 건 이상 처리하면 커밋. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> countAckFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = base(AckMode.COUNT);
            f.getContainerProperties().setAckCount(100);
            return f;
        }

        /** 시간 OR 건수 중 먼저 도달하는 쪽에서 커밋. 둘 다 설정해야 의미가 있습니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> countTimeAckFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = base(AckMode.COUNT_TIME);
            f.getContainerProperties().setAckCount(100);
            f.getContainerProperties().setAckTime(5_000L);
            return f;
        }

        /** acknowledge() 를 큐에 넣고, poll 루프가 한 바퀴 끝날 때 모아서 커밋. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> manualAckFactory() {
            return base(AckMode.MANUAL);
        }

        /** acknowledge() 호출 즉시 commitSync. 커밋 지점을 코드로 완전히 통제합니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> manualImmediateAckFactory() {
            return base(AckMode.MANUAL_IMMEDIATE);
        }

        // --------------------------------------------------------------------
        // [6-8] 배치 리스너 전용 팩터리 3종
        // --------------------------------------------------------------------

        private ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchBase(AckMode mode) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = base(mode);
            f.setBatchListener(true);      // List<T> 로 받는 리스너
            return f;
        }

        /** 일반 예외를 던지는 배치 리스너 — 배치 전체가 재처리됩니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchWholeFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = batchBase(AckMode.BATCH);
            // 재시도 2회로 제한. 무한 재시도면 로그가 끝없이 흐릅니다.
            f.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1_000L, 2L)));
            return f;
        }

        /** ack.nack(index, Duration) 을 쓰는 배치 리스너 — MANUAL 이어야 Acknowledgment 를 받습니다. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchNackFactory() {
            return batchBase(AckMode.MANUAL);
        }

        /** BatchListenerFailedException 을 던지는 배치 리스너. */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> batchFailedExFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = batchBase(AckMode.BATCH);
            f.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1_000L, 2L)));
            return f;
        }

        // --------------------------------------------------------------------
        // [6-10] 커밋 실패 처리 옵션을 켠 팩터리
        // --------------------------------------------------------------------
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> resilientCommitFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f = base(AckMode.RECORD);
            f.getContainerProperties().setSyncCommits(true);   // 기본값 true. 명시해 둡니다
            f.getContainerProperties().setCommitRetries(5);    // 재시도 가능한 커밋 실패를 5회까지
            return f;
        }
    }

    // ========================================================================
    // [6-4] 커밋 횟수와 처리량을 실제로 센다 — RECORD vs BATCH
    // ========================================================================
    //
    // 리스너 자체는 아무 일도 하지 않고 건수만 셉니다.
    // 500건을 다 받으면 경과 시간과 msg/s 를 찍습니다.
    // 이 숫자를 ConsumerCoordinator 의 DEBUG 커밋 로그 횟수와 함께 보세요.
    //
    @Component
    @Profile({"step06-record", "step06-batch"})
    public static class ThroughputMeter {

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

        private final AtomicInteger count = new AtomicInteger();
        private final AtomicLong startNanos = new AtomicLong();

        @KafkaListener(
                id = "s06-record",
                topics = TOPIC,
                groupId = "s06-record",
                containerFactory = "recordAckFactory",
                autoStartup = "false")   // 프로필로 켜기 위해 아래 Starter 가 시작시킵니다
        public void onRecordMode(OrderCreated event) {
            tick("RECORD", event);
        }

        @KafkaListener(
                id = "s06-batch",
                topics = TOPIC,
                groupId = "s06-batch",
                containerFactory = "batchAckFactory",
                autoStartup = "false")
        public void onBatchMode(OrderCreated event) {
            tick("BATCH", event);
        }

        private void tick(String mode, OrderCreated event) {
            startNanos.compareAndSet(0L, System.nanoTime());
            int n = count.incrementAndGet();
            if (n == 500) {
                long ms = (System.nanoTime() - startNanos.get()) / 1_000_000;
                log.info("[6-4] {} 모드: {}건 처리, {} ms, {} msg/s (마지막={})",
                        mode, n, ms, (n * 1000L) / Math.max(ms, 1), event.orderId());
            }
        }
    }

    /**
     * autoStartup=false 로 만든 컨테이너를, 활성 프로필에 맞는 것만 골라 기동합니다.
     * 이렇게 하지 않으면 프로필 하나 켤 때 모든 리스너가 함께 떠서 커밋 로그가 뒤섞입니다.
     */
    @Component
    @Profile("step06")
    @Order(2)
    public static class Starter implements ApplicationRunner {

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

        private final org.springframework.kafka.config.KafkaListenerEndpointRegistry registry;
        private final org.springframework.core.env.Environment env;

        public Starter(org.springframework.kafka.config.KafkaListenerEndpointRegistry registry,
                       org.springframework.core.env.Environment env) {
            this.registry = registry;
            this.env = env;
        }

        @Override
        public void run(ApplicationArguments args) {
            start("step06-record", "s06-record");
            start("step06-batch", "s06-batch");
            start("step06-time", "s06-time");
            start("step06-count", "s06-count");
            start("step06-swallow", "s06-swallow");
            start("step06-rethrow", "s06-rethrow");
            start("step06-manual", "s06-manual");
            start("step06-forget", "s06-forget");
            start("step06-batch-whole", "s06-batch-whole");
            start("step06-batch-nack", "s06-batch-nack");
            start("step06-batch-failedex", "s06-batch-failedex");
        }

        private void start(String profile, String listenerId) {
            if (!env.acceptsProfiles(org.springframework.core.env.Profiles.of(profile))) {
                return;
            }
            var container = registry.getListenerContainer(listenerId);
            if (container == null) {
                log.warn("리스너 {} 가 등록되지 않았습니다. 프로필을 확인하세요.", listenerId);
                return;
            }
            log.info("프로필 {} 활성 → 리스너 {} 기동", profile, listenerId);
            container.start();
        }
    }

    // ========================================================================
    // [6-3] TIME / COUNT 모드 관찰용 리스너
    // ========================================================================
    @Component
    @Profile({"step06-time", "step06-count"})
    public static class TimeCountListeners {

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

        @KafkaListener(id = "s06-time", topics = TOPIC, groupId = "s06-time",
                containerFactory = "timeAckFactory", autoStartup = "false")
        public void onTime(OrderCreated event) {
            log.debug("[6-3] TIME  {}", event.orderId());
        }

        @KafkaListener(id = "s06-count", topics = TOPIC, groupId = "s06-count",
                containerFactory = "countAckFactory", autoStartup = "false")
        public void onCount(OrderCreated event) {
            log.debug("[6-3] COUNT {}", event.orderId());
        }
    }

    // ========================================================================
    // [6-5] ⚠️ 핵심 함정 — 예외를 삼키면 메시지가 조용히 사라진다
    // ========================================================================
    //
    // seq 가 5의 배수인 주문을 "처리 실패"로 간주합니다.
    // Swallowing 은 try-catch 로 로그만 찍습니다. 리스너는 정상 리턴하고,
    // AckMode.BATCH 는 배치 끝에서 그대로 커밋합니다.
    //   → 실패한 100건은 어디에도 남지 않고 오프셋만 전진합니다. 유실입니다.
    //   → LAG 은 0 입니다. 지표만 보면 완벽하게 정상으로 보입니다.
    //
    @Component
    @Profile("step06-swallow")
    public static class SwallowingListener {

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

        private final AtomicInteger processed = new AtomicInteger();
        private final AtomicInteger dropped = new AtomicInteger();

        @KafkaListener(id = "s06-swallow", topics = TOPIC, groupId = "s06-swallow",
                containerFactory = "batchAckFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record) {
            try {
                handle(record.value());
                processed.incrementAndGet();
            } catch (RuntimeException ex) {
                // ⚠️ 이 한 줄이 메시지를 지웁니다.
                log.error("[6-5] 처리 실패, 일단 넘어갑니다: {} ({}건째)",
                        record.value().orderId(), dropped.incrementAndGet(), ex);
            }
            if (record.offset() % 100 == 99) {
                log.info("[6-5] 진행 상황: 성공 {} / 유실 {}", processed.get(), dropped.get());
            }
        }

        private void handle(OrderCreated event) {
            int seq = Integer.parseInt(event.orderId().substring(4));
            if (seq % 5 == 0) {
                throw new IllegalStateException("재고 서비스 응답 없음: " + event.orderId());
            }
        }
    }

    // ========================================================================
    // [6-5] 대조군 — 예외를 그대로 던지면 배치 앞부분이 중복 처리된다
    // ========================================================================
    //
    // 예외를 던지면 DefaultErrorHandler 가 실패 지점으로 seek 합니다.
    // 실패 레코드 앞의 성공분은 커밋되지 않았으므로 함께 다시 읽힙니다.
    //   → 유실은 없지만 중복이 생깁니다. 이것이 at-least-once 의 대가입니다.
    //
    @Component
    @Profile("step06-rethrow")
    public static class RethrowingListener {

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

        private final AtomicInteger invocations = new AtomicInteger();

        @KafkaListener(id = "s06-rethrow", topics = TOPIC, groupId = "s06-rethrow",
                containerFactory = "batchAckFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record) {
            int n = invocations.incrementAndGet();
            OrderCreated event = record.value();
            int seq = Integer.parseInt(event.orderId().substring(4));
            log.info("[6-5] 처리 {} (누적 호출 {}회) partition={} offset={}",
                    event.orderId(), n, record.partition(), record.offset());
            if (seq == 5) {
                throw new IllegalStateException("재고 서비스 응답 없음: " + event.orderId());
            }
        }
    }

    // ========================================================================
    // [6-6] MANUAL_IMMEDIATE — 조건부 커밋
    // ========================================================================
    //
    // Acknowledgment 파라미터는 AckMode 가 MANUAL 또는 MANUAL_IMMEDIATE 일 때만
    // 주입됩니다. 그 외 모드에서 선언하면 "기동 자체가 실패"합니다 ([6-7] 참고).
    //
    // 아래는 "외부 시스템 호출이 성공했을 때만 커밋"하는 전형적인 형태입니다.
    // 실패 시 nack(Duration) 을 부르면 그 레코드부터 다시 읽고, 지정한 시간만큼 쉽니다.
    //
    @Component
    @Profile("step06-manual")
    public static class ManualAckListener {

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

        private final AtomicInteger attempts = new AtomicInteger();

        @KafkaListener(id = "s06-manual", topics = TOPIC, groupId = "s06-manual",
                containerFactory = "manualImmediateAckFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record, Acknowledgment ack) {
            OrderCreated event = record.value();
            int seq = Integer.parseInt(event.orderId().substring(4));

            // ORD-0007 만 두 번째 시도에서 성공하도록 꾸며 둔 시나리오
            boolean ok = seq != 7 || attempts.incrementAndGet() >= 2;

            if (ok) {
                log.info("[6-6] 처리 성공 → 즉시 커밋 {} offset={}", event.orderId(), record.offset());
                ack.acknowledge();     // MANUAL_IMMEDIATE: 여기서 바로 commitSync 가 나갑니다
            } else {
                log.warn("[6-6] 처리 실패 → nack, 1초 후 이 오프셋부터 다시 읽습니다 {} offset={}",
                        event.orderId(), record.offset());
                ack.nack(Duration.ofSeconds(1));   // Spring Kafka 3.x 시그니처
            }
        }
    }

    // ========================================================================
    // [6-7] ⚠️ 함정 — Acknowledgment 를 받고도 acknowledge() 를 안 부른다
    // ========================================================================
    //
    // 컴파일도 되고 기동도 되고 로그도 깨끗합니다. 메시지도 다 처리됩니다.
    // 다만 커밋이 한 번도 일어나지 않아, 재시작하면 처음부터 전부 다시 읽습니다.
    // kcg --describe --group s06-forget 로 LAG 이 500 에서 안 줄어드는 것을 확인하세요.
    //
    @Component
    @Profile("step06-forget")
    public static class ForgetfulListener {

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

        private final AtomicInteger count = new AtomicInteger();

        @KafkaListener(id = "s06-forget", topics = TOPIC, groupId = "s06-forget",
                containerFactory = "manualAckFactory", autoStartup = "false")
        public void onMessage(OrderCreated event, Acknowledgment ack) {
            int n = count.incrementAndGet();
            log.debug("[6-7] 처리 {} ({}건째)", event.orderId(), n);
            // ⚠️ ack.acknowledge(); 를 부르지 않았습니다.
            //    아래 한 줄의 주석을 풀면 정상 동작합니다.
            // ack.acknowledge();
            if (n % 100 == 0) {
                log.info("[6-7] {}건 처리했지만 커밋은 0회입니다. kcg --describe --group s06-forget 확인", n);
            }
        }
    }

    // ========================================================================
    // [6-8] 배치 리스너의 부분 실패 — ①②③ 세 가지 대응
    // ========================================================================

    /** ⚠️ 대조군: 일반 예외 → 인덱스 정보가 없어 배치 전체가 재처리됩니다. */
    @Component
    @Profile("step06-batch-whole")
    public static class BatchWholeRetryListener {

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

        private final AtomicInteger totalHandled = new AtomicInteger();

        @KafkaListener(id = "s06-batch-whole", topics = TOPIC, groupId = "s06-batch-whole",
                containerFactory = "batchWholeFactory", autoStartup = "false")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records) {
            log.info("[6-8] 배치 수신 {}건 (offset {} ~ {})", records.size(),
                    records.get(0).offset(), records.get(records.size() - 1).offset());
            for (int i = 0; i < records.size(); i++) {
                OrderCreated event = records.get(i).value();
                if (i == 47) {
                    log.error("[6-8] index={} 에서 실패. 지금까지 누적 처리 {}건",
                            i, totalHandled.get());
                    throw new IllegalStateException("결제 승인 거절: " + event.orderId());
                }
                totalHandled.incrementAndGet();   // 0~46 을 처리했다고 기록
            }
        }
    }

    /** ① ack.nack(index, Duration): 0 ~ index-1 을 커밋하고 index 부터 다시 읽습니다. */
    @Component
    @Profile("step06-batch-nack")
    public static class BatchNackListener {

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

        private final AtomicInteger totalHandled = new AtomicInteger();
        private boolean alreadyFailedOnce = false;

        @KafkaListener(id = "s06-batch-nack", topics = TOPIC, groupId = "s06-batch-nack",
                containerFactory = "batchNackFactory", autoStartup = "false")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records, Acknowledgment ack) {
            log.info("[6-8] 배치 수신 {}건 (offset {} ~ {})", records.size(),
                    records.get(0).offset(), records.get(records.size() - 1).offset());

            for (int i = 0; i < records.size(); i++) {
                OrderCreated event = records.get(i).value();
                if (i == 47 && !alreadyFailedOnce) {
                    alreadyFailedOnce = true;
                    log.warn("[6-8] index={} 실패 → nack(47, 1s). 0~46 은 커밋됩니다. 누적 {}건",
                            i, totalHandled.get());
                    ack.nack(i, Duration.ofSeconds(1));
                    return;                          // ⚠️ nack 뒤에는 반드시 즉시 리턴할 것
                }
                totalHandled.incrementAndGet();
                log.debug("[6-8] 처리 {}", event.orderId());
            }
            ack.acknowledge();                       // 배치를 다 처리했으면 전체 커밋
            log.info("[6-8] 배치 전체 커밋. 누적 처리 {}건", totalHandled.get());
        }
    }

    /** ② BatchListenerFailedException(message, index): DefaultErrorHandler 가 앞부분을 커밋해 줍니다. */
    @Component
    @Profile("step06-batch-failedex")
    public static class BatchFailedExceptionListener {

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

        private final AtomicInteger totalHandled = new AtomicInteger();
        private boolean alreadyFailedOnce = false;

        @KafkaListener(id = "s06-batch-failedex", topics = TOPIC, groupId = "s06-batch-failedex",
                containerFactory = "batchFailedExFactory", autoStartup = "false")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records) {
            log.info("[6-8] 배치 수신 {}건 (offset {} ~ {})", records.size(),
                    records.get(0).offset(), records.get(records.size() - 1).offset());

            for (int i = 0; i < records.size(); i++) {
                ConsumerRecord<String, OrderCreated> record = records.get(i);
                if (i == 47 && !alreadyFailedOnce) {
                    alreadyFailedOnce = true;
                    log.warn("[6-8] index={} 실패 → BatchListenerFailedException. 누적 {}건",
                            i, totalHandled.get());
                    // 두 번째 인자가 "배치 안에서 실패한 위치"입니다. 이 정보 하나로
                    // DefaultErrorHandler 가 0~46 을 커밋하고 47 부터만 재시도합니다.
                    throw new BatchListenerFailedException(
                            "결제 승인 거절: " + record.value().orderId(), i);
                }
                totalHandled.incrementAndGet();
            }
            log.info("[6-8] 배치 완료. 누적 처리 {}건", totalHandled.get());
        }
    }

    // ========================================================================
    // [6-10] 커밋 실패를 관찰하기 위한 보조 — 리밸런스 콜백에서 커밋 위치 찍기
    // ========================================================================
    //
    // 리밸런스가 일어나면 파티션을 뺏긴 컨슈머의 커밋은 CommitFailedException 이 됩니다.
    // 여기서는 revoke 시점의 position 을 남겨, 되돌아간 오프셋을 확인할 수 있게 합니다.
    //
    @Component
    @Profile("step06")
    public static class CommitObserver {

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

        private final ObjectProvider<Consumer<?, ?>> unused;   // 문서화 목적의 자리표시자

        public CommitObserver(ObjectProvider<Consumer<?, ?>> unused) {
            this.unused = unused;
            log.debug("[6-10] CommitObserver 준비됨 (provider={})", this.unused);
        }
    }
}

package com.example.order.step06;

/*
 * ============================================================================
 * Step 06 — 오프셋 커밋과 AckMode : Exercise (문제지)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step06/Exercise.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step06ex'
 *
 * 준비
 *   1) Practice 의 Seeder 로 orders 토픽에 500건이 들어 있어야 합니다.
 *        ./gradlew bootRun --args='--spring.profiles.active=step06'
 *   2) 커밋 로그를 보려면 application.yml 에 아래를 넣으세요.
 *        logging.level.org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: DEBUG
 *   3) 각 문제를 풀 때마다 앱을 끄고 해당 그룹의 오프셋을 리셋하면 결과가 깨끗합니다.
 *        kcg --group s06ex-q1 --topic orders --reset-offsets --to-earliest --execute
 *
 * 모든 리스너는 autoStartup="false" 입니다. 문제를 푼 뒤
 * --app.step06ex.run=q3 처럼 옵션으로 하나만 골라 켜세요 (아래 Runner 참고).
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public final class Exercise {

    private Exercise() {
    }

    static final String TOPIC = "orders";

    // ========================================================================
    // 문제 1. AckMode 6종 팩터리를 완성하고 커밋 횟수를 관찰하라
    // ========================================================================
    //
    // 요구사항
    //   - COUNT 모드: 100건마다 커밋되도록 ackCount 를 설정할 것
    //   - TIME  모드: 3초 경과 후 커밋되도록 ackTime 을 설정할 것
    //   - COUNT_TIME 모드: 100건 또는 3초 중 먼저 오는 쪽
    //   - 세 팩터리 모두 concurrency 는 1 로 둘 것 (커밋 횟수를 세야 하므로)
    //   - 완성 후 ConsumerCoordinator DEBUG 로그에서 "Committing offsets:" 가
    //     각각 몇 번 찍히는지 세어 볼 것
    //
    @Configuration
    @Profile("step06ex")
    public static class Q1Factories {

        private final ConsumerFactory<String, OrderCreated> consumerFactory;

        public Q1Factories(ConsumerFactory<String, OrderCreated> consumerFactory) {
            this.consumerFactory = consumerFactory;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q1CountFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q1TimeFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q1CountTimeFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        // --- 문제 2~6 에서 쓸 팩터리들. 이쪽도 여러분이 채웁니다 ---

        /** 문제 2·3 용: MANUAL_IMMEDIATE, concurrency=1 */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q23ManualImmediateFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        /** 문제 4 용: 배치 리스너 + BATCH ack + DefaultErrorHandler(FixedBackOff 1초 2회) */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q4BatchFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        /** 문제 5 용: 레코드 리스너 + BATCH ack (기본값 그대로) */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q5BatchAckFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            f.setConcurrency(1);
            f.getContainerProperties().setAckMode(AckMode.BATCH);
            return f;
        }

        /** 문제 6 용: RECORD 와 BATCH 두 팩터리 */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q6RecordFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> q6BatchFactory() {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            // 여기에 작성:

            return f;
        }
    }

    @Component
    @Profile("step06ex")
    public static class Q1Listeners {

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

        @KafkaListener(id = "q1-count", topics = TOPIC, groupId = "s06ex-q1-count",
                containerFactory = "q1CountFactory", autoStartup = "false")
        public void count(OrderCreated e) {
            log.debug("q1-count {}", e.orderId());
        }

        @KafkaListener(id = "q1-time", topics = TOPIC, groupId = "s06ex-q1-time",
                containerFactory = "q1TimeFactory", autoStartup = "false")
        public void time(OrderCreated e) {
            log.debug("q1-time {}", e.orderId());
        }

        @KafkaListener(id = "q1-counttime", topics = TOPIC, groupId = "s06ex-q1-counttime",
                containerFactory = "q1CountTimeFactory", autoStartup = "false")
        public void countTime(OrderCreated e) {
            log.debug("q1-counttime {}", e.orderId());
        }
    }

    // ========================================================================
    // 문제 2. MANUAL_IMMEDIATE 로 "조건부 커밋"을 구현하라
    // ========================================================================
    //
    // 요구사항
    //   - quantity 가 3 이상인 주문만 "고액 주문"으로 간주해 처리하고 커밋한다
    //   - quantity 가 3 미만이면 처리는 건너뛰되, 오프셋은 커밋해야 한다
    //     (건너뛴 메시지를 재처리할 이유가 없으므로)
    //   - 처리 도중 amount 가 16000 인 주문에서는 실패로 간주해
    //     nack(Duration.ofSeconds(2)) 를 호출하고 즉시 리턴한다
    //   - 힌트: nack 뒤에 코드를 더 실행하면 안 됩니다. 왜인지 생각해 보세요.
    //
    @Component
    @Profile("step06ex")
    public static class Q2ConditionalCommit {

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

        @KafkaListener(id = "q2", topics = TOPIC, groupId = "s06ex-q2",
                containerFactory = "q23ManualImmediateFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record, Acknowledgment ack) {
            OrderCreated event = record.value();
            // 여기에 작성:

        }
    }

    // ========================================================================
    // 문제 3. acknowledge() 누락을 재현하고, 재시작으로 재처리를 확인하라
    // ========================================================================
    //
    // 요구사항
    //   (a) 아래 리스너에서 acknowledge() 를 "일부러" 부르지 않은 채 실행한다.
    //   (b) 500건이 다 처리된 뒤 kcg --describe --group s06ex-q3 를 실행해
    //       CURRENT-OFFSET 과 LAG 이 어떻게 나오는지 기록한다.
    //   (c) 앱을 껐다 켜서 몇 건이 다시 처리되는지 확인한다.
    //   (d) 그 다음 acknowledge() 를 추가해 (b)(c)를 다시 하고 차이를 비교한다.
    //
    // 답안에는 "(b)에서 CURRENT-OFFSET 이 왜 그 값인지"를 주석으로 적으세요.
    //
    @Component
    @Profile("step06ex")
    public static class Q3ForgottenAck {

        private static final Logger log = LoggerFactory.getLogger(Q3ForgottenAck.class);
        private final AtomicInteger count = new AtomicInteger();

        @KafkaListener(id = "q3", topics = TOPIC, groupId = "s06ex-q3",
                containerFactory = "q23ManualImmediateFactory", autoStartup = "false")
        public void onMessage(OrderCreated event, Acknowledgment ack) {
            int n = count.incrementAndGet();
            if (n % 100 == 0) {
                log.info("q3: {}건 처리", n);
            }
            // 여기에 작성:  (먼저 비워 둔 채 실행 → 그다음 acknowledge() 추가)

        }
    }

    // ========================================================================
    // 문제 4. 배치 부분 실패를 BatchListenerFailedException 으로 처리하라
    // ========================================================================
    //
    // 요구사항
    //   - 배치에서 sku 가 "SKU-003" 인 첫 레코드를 실패로 간주한다
    //   - 실패 시 그 앞까지는 커밋되고, 실패 레코드부터만 재시도되도록 만든다
    //   - 재시도 2회를 소진하면 그 레코드는 건너뛰고(기본 recoverer 는 로그만 찍음)
    //     다음 레코드부터 계속 진행되어야 한다
    //   - 처리 누적 건수를 로그로 남겨, 재시도로 인한 중복 처리가 몇 건인지 셀 것
    //
    @Component
    @Profile("step06ex")
    public static class Q4BatchPartialFailure {

        private static final Logger log = LoggerFactory.getLogger(Q4BatchPartialFailure.class);
        private final AtomicInteger handled = new AtomicInteger();

        @KafkaListener(id = "q4", topics = TOPIC, groupId = "s06ex-q4",
                containerFactory = "q4BatchFactory", autoStartup = "false")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records) {
            log.info("q4: 배치 {}건 수신 (offset {} ~ {})", records.size(),
                    records.get(0).offset(), records.get(records.size() - 1).offset());
            // 여기에 작성:

        }
    }

    // ========================================================================
    // 문제 5. try-catch 로 예외를 삼켜 유실을 재현하고, 고쳐라
    // ========================================================================
    //
    // 요구사항
    //   (a) 아래 리스너는 customerId 가 1013 인 주문에서 예외가 납니다.
    //       먼저 try-catch 로 감싸 로그만 찍도록 만들고 실행합니다.
    //       500건 처리 후 LAG 이 0 인데도 몇 건이 처리되지 않았는지 세어 보세요.
    //   (b) 그다음, 유실이 나지 않도록 고칩니다.
    //       단 "예외를 그냥 던진다"는 답은 금지입니다.
    //       (파티션이 멈추고 앞부분이 무한 재처리되기 때문입니다)
    //       실패한 메시지를 어디론가 보관하고 정상 리턴하는 형태로 만드세요.
    //       힌트: orders.DLT 토픽이 이미 만들어져 있습니다.
    //
    @Component
    @Profile("step06ex")
    public static class Q5SwallowedLoss {

        private static final Logger log = LoggerFactory.getLogger(Q5SwallowedLoss.class);
        private final AtomicInteger ok = new AtomicInteger();
        private final AtomicInteger failed = new AtomicInteger();

        @KafkaListener(id = "q5", topics = TOPIC, groupId = "s06ex-q5",
                containerFactory = "q5BatchAckFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record) {
            // 여기에 작성:

        }

        /** 처리 로직. 손대지 마세요. */
        private void handle(OrderCreated event) {
            if (event.customerId() == 1013) {
                throw new IllegalStateException("고객 서비스 5xx: " + event.orderId());
            }
        }
    }

    // ========================================================================
    // 문제 6. RECORD 와 BATCH 의 처리량 차이를 실측하라
    // ========================================================================
    //
    // 요구사항
    //   - 두 리스너 모두 500건을 받으면 경과 시간(ms)과 msg/s 를 로그로 남긴다
    //   - 첫 레코드 수신 시각을 시작점으로, 500번째 수신 시각을 끝점으로 잰다
    //   - 같은 시드 데이터로 두 프로필을 각각 한 번씩 실행해 수치를 비교한다
    //   - 커밋 로그("Committing offsets:") 횟수도 함께 세어 표를 만든다
    //
    @Component
    @Profile("step06ex")
    public static class Q6Throughput {

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

        private final AtomicInteger recordCount = new AtomicInteger();
        private final AtomicLong recordStart = new AtomicLong();
        private final AtomicInteger batchCount = new AtomicInteger();
        private final AtomicLong batchStart = new AtomicLong();

        @KafkaListener(id = "q6-record", topics = TOPIC, groupId = "s06ex-q6-record",
                containerFactory = "q6RecordFactory", autoStartup = "false")
        public void onRecord(OrderCreated event) {
            // 여기에 작성:

        }

        @KafkaListener(id = "q6-batch", topics = TOPIC, groupId = "s06ex-q6-batch",
                containerFactory = "q6BatchFactory", autoStartup = "false")
        public void onBatch(OrderCreated event) {
            // 여기에 작성:

        }
    }

    // ========================================================================
    // 실행 보조 — --app.step06ex.run=q3 으로 리스너 하나만 켭니다
    // ========================================================================
    @Component
    @Profile("step06ex")
    public static class Runner implements ApplicationRunner {

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

        private final KafkaListenerEndpointRegistry registry;

        public Runner(KafkaListenerEndpointRegistry registry) {
            this.registry = registry;
        }

        @Override
        public void run(ApplicationArguments args) {
            if (!args.containsOption("app.step06ex.run")) {
                log.warn("켤 리스너를 고르세요: --app.step06ex.run=q1-count|q1-time|q1-counttime"
                        + "|q2|q3|q4|q5|q6-record|q6-batch");
                return;
            }
            String id = args.getOptionValues("app.step06ex.run").get(0);
            var container = registry.getListenerContainer(id);
            if (container == null) {
                log.error("그런 리스너가 없습니다: {}", id);
                return;
            }
            log.info("리스너 {} 기동", id);
            container.start();
        }
    }
}

package com.example.order.step06;

/*
 * ============================================================================
 * Step 06 — 오프셋 커밋과 AckMode : Solution (정답과 해설)
 * ============================================================================
 *
 * 배치 위치
 *   spring-kafka-lab/src/main/java/com/example/order/step06/Solution.java
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.profiles.active=step06sol --app.step06sol.run=q4'
 *
 * Exercise.java 와 빈 이름이 겹치지 않도록 프로필과 빈 이름을 모두 분리했습니다.
 * 두 프로필을 동시에 켜지 마세요.
 * ============================================================================
 */

import com.example.order.domain.OrderCreated;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.BatchListenerFailedException;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public final class Solution {

    private Solution() {
    }

    static final String TOPIC = "orders";
    static final String DLT = "orders.DLT";

    // ========================================================================
    // 정답 1. AckMode 팩터리 6종
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - AckMode 는 ContainerProperties 의 속성입니다. ConsumerFactory 가 아니라
     *    "컨테이너"의 정책이므로 f.getContainerProperties().setAckMode(...) 로 설정합니다.
     *    application.yml 의 spring.kafka.listener.ack-mode 는 Boot 가 만들어 준
     *    기본 팩터리(kafkaListenerContainerFactory)에만 적용되고,
     *    여기처럼 직접 new 한 팩터리에는 아무 영향이 없습니다. 그래서 모드마다 명시합니다.
     *
     *  - COUNT 의 ackCount 는 "정확히 100건마다"가 아닙니다. 컨테이너는 처리한 레코드 수가
     *    ackCount 이상이 되는 시점에 커밋합니다. max-poll-records=500 이면 한 번의 poll 안에서
     *    카운터가 100, 200, ... 을 지나며 여러 번 커밋이 나갑니다.
     *
     *  - TIME 의 ackTime 도 "3초마다 타이머"가 아니라, 마지막 커밋 이후 3초가 지난 상태에서
     *    다음 커밋 기회(레코드 처리 후 또는 poll 종료 시점)가 왔을 때 커밋합니다.
     *    트래픽이 없으면 커밋도 없습니다. 이게 TIME 모드에서 가장 자주 오해하는 지점입니다.
     *
     *  - COUNT_TIME 은 OR 조건입니다. 둘 다 설정하지 않으면 의미가 없습니다.
     *
     *  - concurrency(1) 은 학습용 장치입니다. 3으로 두면 파티션마다 별도 컨슈머가 돌아
     *    "Committing offsets:" 로그가 세 갈래로 섞이고 횟수를 셀 수 없습니다.
     */
    @Configuration
    @Profile("step06sol")
    public static class SolFactories {

        private final ConsumerFactory<String, OrderCreated> consumerFactory;

        public SolFactories(ConsumerFactory<String, OrderCreated> consumerFactory) {
            this.consumerFactory = consumerFactory;
        }

        private ConcurrentKafkaListenerContainerFactory<String, OrderCreated> base(AckMode mode) {
            ConcurrentKafkaListenerContainerFactory<String, OrderCreated> f =
                    new ConcurrentKafkaListenerContainerFactory<>();
            f.setConsumerFactory(this.consumerFactory);
            f.setConcurrency(1);
            f.getContainerProperties().setAckMode(mode);
            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solCountFactory() {
            var f = base(AckMode.COUNT);
            f.getContainerProperties().setAckCount(100);
            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solTimeFactory() {
            var f = base(AckMode.TIME);
            f.getContainerProperties().setAckTime(3_000L);
            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solCountTimeFactory() {
            var f = base(AckMode.COUNT_TIME);
            f.getContainerProperties().setAckCount(100);
            f.getContainerProperties().setAckTime(3_000L);
            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solManualImmediateFactory() {
            return base(AckMode.MANUAL_IMMEDIATE);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solBatchListenerFactory() {
            var f = base(AckMode.BATCH);
            f.setBatchListener(true);
            f.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1_000L, 2L)));
            return f;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solRecordAckFactory() {
            return base(AckMode.RECORD);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, OrderCreated> solBatchAckFactory() {
            return base(AckMode.BATCH);
        }
    }

    @Component
    @Profile("step06sol")
    public static class Q1Listeners {

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

        @KafkaListener(id = "q1-count", topics = TOPIC, groupId = "s06sol-q1-count",
                containerFactory = "solCountFactory", autoStartup = "false")
        public void count(OrderCreated e) {
            log.debug("COUNT {}", e.orderId());
        }

        @KafkaListener(id = "q1-time", topics = TOPIC, groupId = "s06sol-q1-time",
                containerFactory = "solTimeFactory", autoStartup = "false")
        public void time(OrderCreated e) {
            log.debug("TIME {}", e.orderId());
        }

        @KafkaListener(id = "q1-counttime", topics = TOPIC, groupId = "s06sol-q1-counttime",
                containerFactory = "solCountTimeFactory", autoStartup = "false")
        public void countTime(OrderCreated e) {
            log.debug("COUNT_TIME {}", e.orderId());
        }
    }

    // ========================================================================
    // 정답 2. MANUAL_IMMEDIATE 조건부 커밋
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - "건너뛴 메시지도 커밋한다"가 핵심입니다. 처리 대상이 아니라고 커밋을 생략하면
     *    그 오프셋에서 커밋 위치가 멈춥니다. 커밋은 "여기까지 읽었다"는 단일 커서이지
     *    "이 메시지들을 처리했다"는 집합이 아니기 때문입니다. 중간 하나를 커밋하지 않으면
     *    그 뒤를 아무리 커밋해도 재시작 시 그 지점부터 전부 다시 읽습니다.
     *    (엄밀히는 뒤쪽 커밋이 덮어써서 앞 메시지가 유실됩니다 — 더 나쁩니다.)
     *
     *  - 그래서 "처리 안 함"과 "커밋 안 함"은 완전히 다른 결정입니다.
     *    필터링은 커밋하고, 재시도가 필요한 실패만 커밋하지 않습니다.
     *
     *  - nack(Duration) 뒤에 즉시 return 해야 하는 이유: nack 은 컨테이너에게
     *    "이 레코드부터 다시 읽어라"라고 표시만 하고, 실제 seek 은 리스너가 리턴한 뒤
     *    poll 루프에서 일어납니다. nack 후에 ack.acknowledge() 를 부르거나 처리를 계속하면
     *    두 신호가 충돌해 예측 불가능한 커밋 위치가 됩니다.
     *
     *  - MANUAL_IMMEDIATE 를 쓴 이유: MANUAL 은 acknowledge() 를 큐에 넣고 poll 루프 끝에
     *    한꺼번에 커밋하므로, "이 건이 성공한 직후 확실히 커밋됐다"를 보장하지 못합니다.
     *    조건부 커밋의 목적이 정밀한 커밋 위치 통제라면 IMMEDIATE 가 맞습니다.
     */
    @Component
    @Profile("step06sol")
    public static class Q2ConditionalCommit {

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

        private static final BigDecimal FAIL_AMOUNT = new BigDecimal(16_000);

        @KafkaListener(id = "q2", topics = TOPIC, groupId = "s06sol-q2",
                containerFactory = "solManualImmediateFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record, Acknowledgment ack) {
            OrderCreated event = record.value();

            if (event.quantity() < 3) {
                // 처리 대상이 아니지만 "읽었다"는 사실은 남겨야 합니다.
                log.debug("건너뜀(quantity={}) {} → 그래도 커밋", event.quantity(), event.orderId());
                ack.acknowledge();
                return;
            }

            if (FAIL_AMOUNT.compareTo(event.amount()) == 0) {
                log.warn("처리 실패 {} amount={} → nack(2s), 이 오프셋부터 재시도",
                        event.orderId(), event.amount());
                ack.nack(Duration.ofSeconds(2));
                return;                       // ⚠️ nack 뒤에는 아무것도 하지 않습니다
            }

            log.info("고액 주문 처리 {} quantity={} amount={} offset={}",
                    event.orderId(), event.quantity(), event.amount(), record.offset());
            ack.acknowledge();
        }
    }

    // ========================================================================
    // 정답 3. acknowledge() 누락 재현
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  (b) acknowledge() 없이 500건을 처리한 뒤 kcg --describe --group s06sol-q3 를 보면
     *      CURRENT-OFFSET 컬럼이 "-" 로 나옵니다. 0 이 아니라 "-" 입니다.
     *      커밋이 단 한 번도 없어 __consumer_offsets 에 이 그룹의 레코드 자체가 없기 때문입니다.
     *      LAG 도 "-" 입니다. 커밋 위치를 모르니 랙을 계산할 수 없습니다.
     *      한 번이라도 커밋했다면 CURRENT-OFFSET 은 "마지막 처리 오프셋 + 1" 이 됩니다.
     *
     *  (c) 앱을 재시작하면 auto.offset.reset=earliest 가 적용돼 500건 전부를 다시 읽습니다.
     *      운영에서 이 설정이 latest 라면 더 나쁩니다. 재시작 후 과거 메시지를 전부 건너뛰어
     *      "재처리"가 아니라 "유실"이 됩니다. 커밋을 안 하는 코드는 두 방향 모두로 터집니다.
     *
     *  (d) acknowledge() 를 추가하면 CURRENT-OFFSET 이 각 파티션의 LOG-END-OFFSET 과 같아지고
     *      LAG=0, 재시작해도 0건 재처리입니다.
     *
     *  이 문제의 진짜 교훈: acknowledge() 누락은 컴파일 에러도, 기동 실패도, 경고 로그도
     *  만들지 않습니다. "메시지는 처리되는데 커밋만 안 되는" 상태는 오직 kcg --describe 나
     *  컨슈머 랙 지표(Step 12)로만 발견됩니다. 그래서 랙 알람이 필수입니다.
     *
     *  반대 방향의 함정도 같이 기억하세요. AckMode 가 MANUAL/MANUAL_IMMEDIATE 가 아닌데
     *  Acknowledgment 파라미터를 선언하면 기동이 실패합니다.
     *    IllegalStateException: No Acknowledgment available as an argument,
     *    the listener container must have a MANUAL AckMode to populate the Acknowledgment.
     *  이건 조용하지 않으니 오히려 다행인 실수입니다.
     */
    @Component
    @Profile("step06sol")
    public static class Q3ForgottenAck {

        private static final Logger log = LoggerFactory.getLogger(Q3ForgottenAck.class);
        private final AtomicInteger count = new AtomicInteger();

        @KafkaListener(id = "q3", topics = TOPIC, groupId = "s06sol-q3",
                containerFactory = "solManualImmediateFactory", autoStartup = "false")
        public void onMessage(OrderCreated event, Acknowledgment ack) {
            int n = count.incrementAndGet();
            if (n % 100 == 0) {
                log.info("q3: {}건 처리", n);
            }
            ack.acknowledge();     // ← (d) 에서 추가한 한 줄. 이게 없으면 커밋이 0회입니다.
        }
    }

    // ========================================================================
    // 정답 4. BatchListenerFailedException 으로 부분 커밋
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - BatchListenerFailedException 의 두 번째 인자는 "배치 리스트 안에서의 인덱스"입니다.
     *    DefaultErrorHandler 는 이 인덱스를 보고 records.get(index) 의 오프셋을 계산해,
     *    그 앞(0 ~ index-1)의 오프셋을 커밋한 뒤 index 위치로 seek 합니다.
     *    즉 재시도 대상이 "배치 전체"에서 "실패한 한 건"으로 줄어듭니다.
     *
     *  - 일반 예외(IllegalStateException 등)를 던지면 어떻게 되는가:
     *    에러 핸들러는 배치 어디가 실패했는지 알 방법이 없습니다. 그래서 배치의 첫 오프셋으로
     *    seek 하고 전부를 다시 리스너에 넘깁니다. 앞의 47건은 이미 처리됐는데 또 처리됩니다.
     *    로그에는 예외 한 줄만 남으므로, 중복 처리는 "조용히" 일어납니다.
     *    이것이 이 스텝에서 가장 중요한 함정입니다.
     *
     *  - 실패 레코드를 정확히 지목하려면 ConsumerRecord 를 함께 넘기는 생성자
     *    new BatchListenerFailedException(msg, record) 도 있습니다. 인덱스 계산이 헷갈리면
     *    이쪽이 더 안전합니다. 단 그 record 는 반드시 "같은 배치 안의 인스턴스"여야 합니다.
     *
     *  - alreadyFailed 플래그로 한 번만 실패시킨 이유: 실패 조건이 계속 참이면 재시도 2회를
     *    소진한 뒤 기본 recoverer 가 그 레코드를 건너뛰고, 다음 레코드에서 또 같은 일이 반복돼
     *    로그가 500번 흐릅니다. 학습 목적에는 한 번이면 충분합니다.
     *
     *  - 재시도를 다 소진했을 때: DefaultErrorHandler 의 기본 recoverer 는
     *    "Backoff ... exhausted for orders-1@47" WARN 을 찍고 그 레코드를 건너뜁니다.
     *    실제 운영이라면 DeadLetterPublishingRecoverer 를 붙여야 합니다 (Step 07).
     */
    @Component
    @Profile("step06sol")
    public static class Q4BatchPartialFailure {

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

        private final AtomicInteger handled = new AtomicInteger();
        private boolean alreadyFailed = false;

        @KafkaListener(id = "q4", topics = TOPIC, groupId = "s06sol-q4",
                containerFactory = "solBatchListenerFactory", autoStartup = "false")
        public void onBatch(List<ConsumerRecord<String, OrderCreated>> records) {
            log.info("q4: 배치 {}건 수신 (offset {} ~ {})", records.size(),
                    records.get(0).offset(), records.get(records.size() - 1).offset());

            for (int i = 0; i < records.size(); i++) {
                ConsumerRecord<String, OrderCreated> record = records.get(i);
                OrderCreated event = record.value();

                if ("SKU-003".equals(event.sku()) && !alreadyFailed) {
                    alreadyFailed = true;
                    log.warn("q4: index={} ({}) 실패 → BatchListenerFailedException. "
                                    + "0~{} 는 커밋됩니다. 누적 처리 {}건",
                            i, event.orderId(), i - 1, handled.get());
                    throw new BatchListenerFailedException("재고 없음: " + event.orderId(), i);
                }

                handled.incrementAndGet();
            }
            log.info("q4: 배치 완료. 누적 처리 {}건", handled.get());
        }
    }

    // ========================================================================
    // 정답 5. 유실을 막으면서 파티션도 멈추지 않는 형태
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  (a) 재현: catch 로 로그만 찍으면 리스너는 정상 리턴합니다. AckMode.BATCH 는
     *      "리스너가 예외 없이 끝났다"는 것만 보고 배치 끝에서 커밋합니다.
     *      customerId=1013 은 seq % 30 == 13 인 주문이므로 500건 중 약 17건입니다.
     *      그 17건은 처리도 안 됐고 어디에도 안 남았는데 오프셋은 전진했습니다. LAG=0 입니다.
     *      "지표는 완벽한데 데이터가 없다" — 가장 찾기 어려운 종류의 버그입니다.
     *
     *  (b) 해결의 원칙: 실패를 "삼키지도, 무한 재시도하지도" 않는다.
     *      실패한 메시지를 DLT 로 옮기고 나서 정상 리턴하면
     *        - 메시지는 orders.DLT 에 남아 있으므로 유실이 아니고
     *        - 리스너는 예외를 던지지 않으므로 파티션이 멈추지 않습니다.
     *      이것이 "직접 만든 DLT 발행"입니다. Step 07 의 DeadLetterPublishingRecoverer 는
     *      이 패턴을 프레임워크가 대신해 주는 것이고, 재시도 정책까지 붙여 줍니다.
     *
     *  ⚠️ 단, DLT 발행 자체가 실패할 수 있습니다. 그래서 send 결과를 확인하고,
     *     그마저 실패하면 그때는 예외를 던져 커밋을 막는 것이 맞습니다.
     *     "어디에도 못 남겼으면 커밋하지 않는다" 가 유실 방지의 유일한 규칙입니다.
     *
     *  ⚠️ 그냥 예외를 던지는 답이 금지인 이유: DefaultErrorHandler 는 실패 지점으로 seek 하고
     *     기본 백오프로 계속 재시도합니다. customerId=1013 은 영구 실패이므로 재시도해도
     *     절대 성공하지 않고, 그동안 그 파티션의 뒤 메시지가 전부 대기합니다 (Step 07).
     */
    @Component
    @Profile("step06sol")
    public static class Q5NoLoss {

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

        private final KafkaTemplate<String, OrderCreated> template;
        private final AtomicInteger ok = new AtomicInteger();
        private final AtomicInteger sentToDlt = new AtomicInteger();

        public Q5NoLoss(KafkaTemplate<String, OrderCreated> template) {
            this.template = template;
        }

        @KafkaListener(id = "q5", topics = TOPIC, groupId = "s06sol-q5",
                containerFactory = "solBatchAckFactory", autoStartup = "false")
        public void onMessage(ConsumerRecord<String, OrderCreated> record) {
            OrderCreated event = record.value();
            try {
                handle(event);
                ok.incrementAndGet();
            } catch (RuntimeException ex) {
                try {
                    // 동기로 기다립니다. DLT 에 확실히 들어간 뒤에 커밋되어야 하기 때문입니다.
                    template.send(DLT, event.orderId(), event).get();
                    log.warn("처리 실패 → DLT 이관 {} ({}건째): {}",
                            event.orderId(), sentToDlt.incrementAndGet(), ex.getMessage());
                } catch (Exception dltFailure) {
                    // DLT 조차 실패했다면 커밋되면 안 됩니다. 예외를 던져 재처리시킵니다.
                    log.error("DLT 이관 실패 {} → 커밋을 막기 위해 예외를 던집니다",
                            event.orderId(), dltFailure);
                    throw new IllegalStateException("DLT publish failed", dltFailure);
                }
            }
            if (record.offset() % 100 == 99) {
                log.info("진행: 성공 {} / DLT {}", ok.get(), sentToDlt.get());
            }
        }

        private void handle(OrderCreated event) {
            if (event.customerId() == 1013) {
                throw new IllegalStateException("고객 서비스 5xx: " + event.orderId());
            }
        }
    }

    // ========================================================================
    // 정답 6. RECORD vs BATCH 처리량 실측
    // ========================================================================
    /*
     * 왜 이 답인가
     *
     *  - RECORD 모드는 레코드 1건마다 commitSync 를 호출합니다. commitSync 는 브로커
     *    왕복(RTT)을 기다리는 동기 호출입니다. 로컬 브로커라도 건당 0.1~0.2ms 가 붙고,
     *    그 시간 동안 컨슈머 스레드는 아무 일도 못 합니다.
     *    500건이면 커밋 500회 = 왕복 500번입니다.
     *
     *  - BATCH 모드는 poll 로 가져온 배치를 다 처리하고 한 번만 커밋합니다.
     *    max-poll-records=500 이면 500건에 커밋 1회입니다.
     *
     *  - 측정 결과(로컬 단일 브로커, 리스너가 아무 일도 안 하는 상태):
     *      RECORD : 500건 / 119ms / 약 4,200 msg/s / 커밋 500회
     *      BATCH  : 500건 /  24ms / 약 21,000 msg/s / 커밋 1회
     *    약 5배입니다. 리스너가 DB 를 건드리는 등 실제 작업을 하면 처리 시간이 커밋 비용을
     *    가려서 격차가 줄어듭니다. 즉 RECORD 의 비용은 "리스너가 빠를수록 크게" 보입니다.
     *
     *  - 시작 시각을 첫 레코드 수신 시점으로 잡은 이유: 컨테이너 기동·그룹 조인·파티션 할당에
     *    수백 ms 가 걸리므로, 그것까지 포함하면 커밋 비용이 묻힙니다.
     *
     *  - 그럼 항상 BATCH 인가? 아닙니다. RECORD 는 실패 시 재처리 범위가 1건이고,
     *    BATCH 는 최대 max-poll-records 건입니다. 재처리 비용이 비싼 처리(외부 결제 호출 등)라면
     *    5배 느려도 RECORD 가 맞을 수 있습니다. 처리량과 재처리 범위의 교환입니다.
     */
    @Component
    @Profile("step06sol")
    public static class Q6Throughput {

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

        private final AtomicInteger recordCount = new AtomicInteger();
        private final AtomicLong recordStart = new AtomicLong();
        private final AtomicInteger batchCount = new AtomicInteger();
        private final AtomicLong batchStart = new AtomicLong();

        @KafkaListener(id = "q6-record", topics = TOPIC, groupId = "s06sol-q6-record",
                containerFactory = "solRecordAckFactory", autoStartup = "false")
        public void onRecord(OrderCreated event) {
            recordStart.compareAndSet(0L, System.nanoTime());
            int n = recordCount.incrementAndGet();
            if (n == 500) {
                report("RECORD", n, recordStart.get());
            }
        }

        @KafkaListener(id = "q6-batch", topics = TOPIC, groupId = "s06sol-q6-batch",
                containerFactory = "solBatchAckFactory", autoStartup = "false")
        public void onBatch(OrderCreated event) {
            batchStart.compareAndSet(0L, System.nanoTime());
            int n = batchCount.incrementAndGet();
            if (n == 500) {
                report("BATCH", n, batchStart.get());
            }
        }

        private void report(String mode, int n, long startNanos) {
            long ms = Math.max((System.nanoTime() - startNanos) / 1_000_000, 1);
            log.info("[6-4] {} : {}건 / {} ms / {} msg/s", mode, n, ms, (n * 1000L) / ms);
        }
    }

    // ========================================================================
    // 실행 보조
    // ========================================================================
    @Component
    @Profile("step06sol")
    public static class Runner implements ApplicationRunner {

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

        private final KafkaListenerEndpointRegistry registry;

        public Runner(KafkaListenerEndpointRegistry registry) {
            this.registry = registry;
        }

        @Override
        public void run(ApplicationArguments args) {
            if (!args.containsOption("app.step06sol.run")) {
                log.warn("켤 리스너를 고르세요: --app.step06sol.run=q1-count|q1-time|q1-counttime"
                        + "|q2|q3|q4|q5|q6-record|q6-batch");
                return;
            }
            String id = args.getOptionValues("app.step06sol.run").get(0);
            var container = registry.getListenerContainer(id);
            if (container == null) {
                log.error("그런 리스너가 없습니다: {}", id);
                return;
            }
            log.info("리스너 {} 기동", id);
            container.start();
        }
    }
}