#Step 07 — ItemProcessor

@FunctionalInterface
public interface ItemProcessor<I, O> {
    @Nullable
    O process(@NonNull I item) throws Exception;
}

package com.example.batch.step07;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.item.ItemProcessor;

import java.math.BigDecimal;
import java.math.RoundingMode;

public class FlatRateSettlementProcessor implements ItemProcessor<Order, Settlement> {

    private static final BigDecimal FLAT_RATE = new BigDecimal("0.0300");

    @Override
    public Settlement process(Order order) {
        BigDecimal gross = order.amount();
        BigDecimal fee   = gross.multiply(FLAT_RATE).setScale(2, RoundingMode.HALF_UP);
        BigDecimal net   = gross.subtract(fee);

        return new Settlement(
                order.order_id(),
                order.customerId(),
                order.orderedAt().toLocalDate(),
                gross,
                FLAT_RATE,
                fee,
                net
        );
    }
}

BigDecimal gross = new BigDecimal("1100.00");
BigDecimal rate  = new BigDecimal("0.0300");
System.out.println(gross.multiply(rate));                              // 스케일 그대로
System.out.println(gross.multiply(rate).setScale(2, RoundingMode.HALF_UP));

new BigDecimal("1050.00").multiply(new BigDecimal("0.0250")).setScale(2);

// ❌ 하지 마세요
@Override
public Settlement process(Order order) {
    BigDecimal rate = jdbcTemplate.queryForObject(
            "SELECT fee_rate FROM customers WHERE customer_id = ?",
            BigDecimal.class, order.customerId());
    // ...
}

package com.example.batch.step07;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.jdbc.core.JdbcTemplate;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.Map;

public class GradeFeeSettlementProcessor implements ItemProcessor<Order, Settlement> {

    /** customer_id → fee_rate. 생성자에서 한 번만 채우고, 이후 읽기 전용입니다. */
    private final Map<Integer, BigDecimal> feeRateByCustomer;

    public GradeFeeSettlementProcessor(JdbcTemplate jdbcTemplate) {
        Map<Integer, BigDecimal> map = new HashMap<>(1400);
        jdbcTemplate.query("SELECT customer_id, fee_rate FROM customers",
                rs -> { map.put(rs.getInt("customer_id"), rs.getBigDecimal("fee_rate")); });
        this.feeRateByCustomer = Map.copyOf(map);   // 불변화 — 7-10 의 무상태 원칙
    }

    @Override
    public Settlement process(Order order) {
        BigDecimal rate = feeRateByCustomer.get(order.customerId());
        if (rate == null) {
            throw new IllegalStateException("등급 정보 없는 고객: " + order.customerId());
        }
        BigDecimal gross = order.amount();
        BigDecimal fee   = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);

        return new Settlement(
                order.order_id(), order.customerId(), order.orderedAt().toLocalDate(),
                gross, rate, fee, gross.subtract(fee));
    }
}

public class CompletedOnlyProcessor implements ItemProcessor<Order, Order> {

    @Override
    public Order process(Order order) {
        if (!"COMPLETED".equals(order.status())) {
            return null;          // ← 이 아이템은 writer 로 내려가지 않습니다
        }
        return order;
    }
}

@Bean
public Step filteringStep(JobRepository jobRepository, PlatformTransactionManager tx,
                          JdbcTemplate jdbcTemplate, DataSource dataSource) {
    return new StepBuilder("filteringStep", jobRepository)
            .<Order, Settlement>chunk(1000, tx)
            .reader(allOrdersReader(dataSource))          // 10만 건 전부
            .processor(new CompositeItemProcessor<>(      // 7-6 에서 설명
                    List.of(new CompletedOnlyProcessor(),
                            new GradeFeeSettlementProcessor(jdbcTemplate))))
            .writer(settlementWriter(dataSource))
            .build();
}

@Bean
public StepExecutionListener countLogger() {
    return new StepExecutionListener() {
        @Override
        public ExitStatus afterStep(StepExecution se) {
            log.info("readCount={}, writeCount={}, filterCount={}, skipCount={}",
                    se.getReadCount(), se.getWriteCount(),
                    se.getFilterCount(), se.getSkipCount());
            return se.getExitStatus();
        }
    };
}

.faultTolerant()
    .skip(IllegalStateException.class)
    .skipLimit(10)

@Bean
public ItemProcessor<Order, Settlement> settlementPipeline(JdbcTemplate jdbcTemplate) {
    CompositeItemProcessor<Order, Settlement> composite = new CompositeItemProcessor<>();
    composite.setDelegates(List.of(
            new CompletedOnlyProcessor(),                    // Order → Order
            new MinAmountFilterProcessor(new BigDecimal("1000")),  // Order → Order
            new GradeFeeSettlementProcessor(jdbcTemplate)    // Order → Settlement
    ));
    return composite;
}

return new CompositeItemProcessor<>(
        new CompletedOnlyProcessor(),
        new MinAmountFilterProcessor(new BigDecimal("1000")),
        new GradeFeeSettlementProcessor(jdbcTemplate));

public class CompositeItemProcessor<I, O> implements ItemProcessor<I, O>, InitializingBean {
    private List<? extends ItemProcessor<?, ?>> delegates;   // ← ?, ? 입니다
    public void setDelegates(List<? extends ItemProcessor<?, ?>> delegates) { ... }
}

// ❌ 순서가 뒤집혔습니다. 그런데 컴파일은 됩니다.
CompositeItemProcessor<Order, Settlement> composite = new CompositeItemProcessor<>();
composite.setDelegates(List.of(
        new GradeFeeSettlementProcessor(jdbcTemplate),   // Order → Settlement
        new CompletedOnlyProcessor()                     // Order 를 기대하는데 Settlement 이 온다
));

@SuppressWarnings("unchecked")
public O process(I item) throws Exception {
    Object result = item;
    for (ItemProcessor<?, ?> delegate : delegates) {
        if (result == null) return null;
        result = processItem((ItemProcessor<Object, Object>) delegate, result);
    }
    return (O) result;   // ← 여기와 델리게이트 안쪽에서 캐스팅이 일어난다
}

@Bean
public ItemProcessor<Order, Settlement> classifyingProcessor(JdbcTemplate jdbcTemplate) {

    ItemProcessor<Order, Settlement> standard = new GradeFeeSettlementProcessor(jdbcTemplate);
    ItemProcessor<Order, Settlement> vip      = new VipSettlementProcessor(jdbcTemplate);
    ItemProcessor<Order, Settlement> drop     = order -> null;   // 정산 대상 아님

    Map<Integer, BigDecimal> rates = loadFeeRates(jdbcTemplate);

    ClassifierCompositeItemProcessor<Order, Settlement> processor =
            new ClassifierCompositeItemProcessor<>();

    processor.setClassifier((Classifier<Order, ItemProcessor<?, ? extends Settlement>>) order -> {
        if (!"COMPLETED".equals(order.status())) return drop;
        BigDecimal rate = rates.get(order.customerId());
        return new BigDecimal("0.0200").compareTo(rate) == 0 ? vip : standard;
    });

    return processor;
}

@FunctionalInterface
public interface Classifier<C, T> extends Serializable {
    T classify(C classifiable);
}

ValidatingItemProcessor<Order> validator = new ValidatingItemProcessor<>(order -> {
    if (order.amount() == null || order.amount().signum() <= 0) {
        throw new ValidationException("금액이 0 이하: order_id=" + order.order_id());
    }
    if (order.orderedAt() == null) {
        throw new ValidationException("주문일시 없음: order_id=" + order.order_id());
    }
});
validator.setFilter(false);   // 기본값. 위반 시 ValidationException 을 던진다

public record Settlement(
        @NotNull Long orderId,
        @NotNull Integer customerId,
        @NotNull LocalDate settleDate,
        @NotNull @DecimalMin("0.00") BigDecimal grossAmount,
        @NotNull @DecimalMin("0.0000") @DecimalMax("0.1000") BigDecimal feeRate,
        @NotNull @DecimalMin("0.00") BigDecimal feeAmount,
        @NotNull @DecimalMin("0.00") BigDecimal netAmount
) {}

@Bean
public BeanValidatingItemProcessor<Settlement> beanValidator() throws Exception {
    BeanValidatingItemProcessor<Settlement> p = new BeanValidatingItemProcessor<>();
    p.setFilter(false);
    p.afterPropertiesSet();     // ← 잊지 마세요
    return p;
}

// ❌ 상태를 가진 프로세서
public class CountingProcessor implements ItemProcessor<Order, Settlement> {
    private int processed = 0;              // ← 상태
    private BigDecimal runningTotal = BigDecimal.ZERO;   // ← 상태

    @Override
    public Settlement process(Order order) {
        processed++;
        runningTotal = runningTotal.add(order.amount());
        // ...
    }
}

.<Order, Settlement>chunk(1000, tx)
.reader(syncReader)
.processor(new CountingProcessor())     // 인스턴스 하나를 4스레드가 공유
.writer(writer)
.taskExecutor(new SimpleAsyncTaskExecutor("batch-"))

package com.example.batch.step07;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import jakarta.validation.constraints.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.support.ClassifierCompositeItemProcessor;
import org.springframework.batch.item.support.CompositeItemProcessor;
import org.springframework.batch.item.validator.BeanValidatingItemProcessor;
import org.springframework.batch.item.validator.ValidatingItemProcessor;
import org.springframework.batch.item.validator.ValidationException;
import org.springframework.classify.Classifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Step 07 — ItemProcessor 실습 전체 코드.
 *
 * 실행:
 *   ./gradlew bootRun --args='--spring.batch.job.name=filteringJob'
 *   ./gradlew bootRun --args='--spring.batch.job.name=classifyingJob'
 *
 * 초기화(실행 전마다):
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 */
@Configuration
public class Practice {

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

    // ======================================================================
    // [7-2] Order -> Settlement 변환과 BigDecimal 반올림 규칙
    // ======================================================================

    /**
     * 수수료율을 상수로 고정한 최소 형태. 변환의 뼈대와 반올림 규칙만 보여 줍니다.
     * 실제 Job 이 쓰는 것은 [7-3] 의 GradeFeeSettlementProcessor 입니다.
     */
    public static class FlatRateSettlementProcessor implements ItemProcessor<Order, Settlement> {

        private static final BigDecimal FLAT_RATE = new BigDecimal("0.0300");

        @Override
        public Settlement process(Order order) {
            BigDecimal gross = order.amount();

            // 규칙: fee 를 먼저 2자리로 "확정" 하고, net 은 확정된 fee 로 뺀다.
            // multiply 는 스케일이 더해지므로(2 + 4 = 6) 여기서 잘라 두지 않으면
            // net 도 6자리가 되고, DB 가 행마다 반올림해 SUM 이 어긋난다.
            BigDecimal fee = gross.multiply(FLAT_RATE).setScale(2, RoundingMode.HALF_UP);
            BigDecimal net = gross.subtract(fee);

            return new Settlement(
                    order.order_id(),
                    order.customerId(),
                    order.orderedAt().toLocalDate(),
                    gross,
                    FLAT_RATE,
                    fee,
                    net);
        }
    }

    /** [7-2] 반올림 동작을 눈으로 확인하는 보조 메서드. main 에서 호출해도 됩니다. */
    public static void demonstrateRounding() {
        BigDecimal gross = new BigDecimal("1100.00");
        BigDecimal rate = new BigDecimal("0.0300");

        System.out.println(gross.multiply(rate));                                // 33.000000
        System.out.println(gross.multiply(rate).setScale(2, RoundingMode.HALF_UP)); // 33.00

        try {
            // setScale(int) 단독은 RoundingMode.UNNECESSARY 입니다.
            new BigDecimal("1050.00").multiply(new BigDecimal("0.0250")).setScale(2);
        } catch (ArithmeticException e) {
            System.out.println("ArithmeticException: " + e.getMessage()); // Rounding necessary
        }
    }

    // ======================================================================
    // [7-3] 등급별 수수료율 — 7만 번 조회를 1번으로
    // ======================================================================

    /**
     * ❌ 나쁜 예. 아이템마다 SELECT 를 날립니다. 7만 건이면 7만 왕복(약 41.9초).
     * 41.9초 -> 5.2초 비교를 재현하고 싶을 때만 파이프라인에 잠깐 끼워 넣으세요.
     */
    public static class SlowLookupProcessor implements ItemProcessor<Order, Settlement> {

        private final JdbcTemplate jdbcTemplate;

        public SlowLookupProcessor(JdbcTemplate jdbcTemplate) {
            this.jdbcTemplate = jdbcTemplate;
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal rate = jdbcTemplate.queryForObject(
                    "SELECT fee_rate FROM customers WHERE customer_id = ?",
                    BigDecimal.class, order.customerId());
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(), gross, rate, fee, gross.subtract(fee));
        }
    }

    /** ✅ 좋은 예. 1,000행짜리 마스터를 생성자에서 한 번만 읽어 불변 맵으로 들고 있습니다. */
    public static class GradeFeeSettlementProcessor implements ItemProcessor<Order, Settlement> {

        // final + 불변 컬렉션 = [7-10] 의 무상태 원칙
        private final Map<Integer, BigDecimal> feeRateByCustomer;

        public GradeFeeSettlementProcessor(JdbcTemplate jdbcTemplate) {
            this.feeRateByCustomer = loadFeeRates(jdbcTemplate);
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal rate = feeRateByCustomer.get(order.customerId());
            if (rate == null) {
                // 데이터가 진짜로 잘못된 경우 -> 예외. null 필터링과 구분합니다([7-4]).
                throw new IllegalStateException("등급 정보 없는 고객: " + order.customerId());
            }
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(), gross, rate, fee, gross.subtract(fee));
        }
    }

    /** VIP 전용. 표준 수수료율에서 0.0050 을 추가 할인합니다. ([7-8] 분기용) */
    public static class VipSettlementProcessor implements ItemProcessor<Order, Settlement> {

        private static final BigDecimal VIP_DISCOUNT = new BigDecimal("0.0050");
        private final Map<Integer, BigDecimal> feeRateByCustomer;

        public VipSettlementProcessor(JdbcTemplate jdbcTemplate) {
            this.feeRateByCustomer = loadFeeRates(jdbcTemplate);
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal base = feeRateByCustomer.get(order.customerId());
            BigDecimal rate = base.subtract(VIP_DISCOUNT);      // 0.0200 -> 0.0150
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(), gross, rate, fee, gross.subtract(fee));
        }
    }

    /** customers 1,000행을 통째로 읽어 불변 맵으로 만듭니다. */
    static Map<Integer, BigDecimal> loadFeeRates(JdbcTemplate jdbcTemplate) {
        Map<Integer, BigDecimal> map = new HashMap<>(1400);
        jdbcTemplate.query("SELECT customer_id, fee_rate FROM customers",
                rs -> { map.put(rs.getInt("customer_id"), rs.getBigDecimal("fee_rate")); });
        return Map.copyOf(map);
    }

    // ======================================================================
    // [7-4] null 반환 = 필터링
    // ======================================================================

    /** COMPLETED 가 아니면 null 을 반환합니다. 예외가 아니라 null 입니다. */
    public static class CompletedOnlyProcessor implements ItemProcessor<Order, Order> {

        @Override
        public Order process(Order order) {
            if (!"COMPLETED".equals(order.status())) {
                return null;    // writer 로 내려가지 않는다 -> filterCount 증가
            }
            return order;
        }
    }

    /** 최소 금액 미만 필터. CompositeItemProcessor 중간 단계 예시입니다. */
    public static class MinAmountFilterProcessor implements ItemProcessor<Order, Order> {

        private final BigDecimal minAmount;

        public MinAmountFilterProcessor(BigDecimal minAmount) {
            this.minAmount = minAmount;
        }

        @Override
        public Order process(Order order) {
            return order.amount().compareTo(minAmount) < 0 ? null : order;
        }
    }

    /** [7-4] StepExecution 의 카운터를 그대로 찍는 리스너. */
    @Bean
    public StepExecutionListener countLogger() {
        return new StepExecutionListener() {
            @Override
            public ExitStatus afterStep(StepExecution se) {
                log.info("readCount={}, writeCount={}, filterCount={}, skipCount={}",
                        se.getReadCount(), se.getWriteCount(),
                        se.getFilterCount(), se.getSkipCount());
                log.info("commitCount={}, rollbackCount={}, processSkipCount={}",
                        se.getCommitCount(), se.getRollbackCount(), se.getProcessSkipCount());
                return se.getExitStatus();
            }
        };
    }

    // ======================================================================
    // [7-6] CompositeItemProcessor — 직렬 파이프라인
    // ======================================================================

    /**
     * ✅ 올바른 순서: 필터를 앞으로, 비싼 변환을 뒤로.
     * 30,000건이 1단계에서 걸러지므로 GradeFee... 는 70,000번만 돕니다.
     */
    @Bean
    public ItemProcessor<Order, Settlement> settlementPipeline(JdbcTemplate jdbcTemplate) {
        // 타입 사슬을 지역 변수로 명시해 두면 눈으로 검증할 수 있습니다([7-7] 방어책 1).
        ItemProcessor<Order, Order> step1 = new CompletedOnlyProcessor();
        ItemProcessor<Order, Order> step2 = new MinAmountFilterProcessor(new BigDecimal("1000"));
        ItemProcessor<Order, Settlement> step3 = new GradeFeeSettlementProcessor(jdbcTemplate);
        // Order -> Order -> Order -> Settlement

        // 5.1 부터는 가변인자 생성자도 제공됩니다.
        return new CompositeItemProcessor<>(step1, step2, step3);
    }

    // ======================================================================
    // [7-7] 제네릭 타입 불일치 — 컴파일은 되는데 런타임에 터진다
    // ======================================================================

    /**
     * ❌ 델리게이트 순서가 뒤집혔습니다. setDelegates 의 파라미터가
     *    List&lt;? extends ItemProcessor&lt;?, ?&gt;&gt; 라서 컴파일러가 아무 말도 하지 않습니다.
     *
     * 실행하면:
     *   java.lang.ClassCastException: class com.example.batch.domain.Settlement
     *       cannot be cast to class com.example.batch.domain.Order
     *       at com.example.batch.step07.Practice$CompletedOnlyProcessor.process(...)
     *
     * 스택트레이스는 CompletedOnlyProcessor 를 가리키지만, 틀린 곳은 아래 setDelegates 입니다.
     *
     * Job 빈으로 등록하지 않았으므로 그냥 빌드해도 안전합니다.
     * 재현하려면 filteringStep 의 .processor(...) 자리에 손으로 바꿔 끼우세요.
     */
    public ItemProcessor<Order, Settlement> brokenPipeline(JdbcTemplate jdbcTemplate) {
        CompositeItemProcessor<Order, Settlement> composite = new CompositeItemProcessor<>();
        composite.setDelegates(List.of(
                new GradeFeeSettlementProcessor(jdbcTemplate),  // Order -> Settlement
                new CompletedOnlyProcessor()                    // Order 를 기대 -> 폭발
        ));
        return composite;
    }

    /**
     * ✅ 방어책 2 — 직접 합성. null 전파를 손으로 써야 하지만,
     *    순서를 틀리면 "incompatible types" 로 컴파일이 실패합니다.
     */
    public ItemProcessor<Order, Settlement> safePipeline(JdbcTemplate jdbcTemplate) {
        ItemProcessor<Order, Order> step1 = new CompletedOnlyProcessor();
        ItemProcessor<Order, Order> step2 = new MinAmountFilterProcessor(new BigDecimal("1000"));
        ItemProcessor<Order, Settlement> step3 = new GradeFeeSettlementProcessor(jdbcTemplate);

        return order -> {
            Order a = step1.process(order);
            if (a == null) return null;
            Order b = step2.process(a);
            if (b == null) return null;
            return step3.process(b);
        };
    }

    // ======================================================================
    // [7-8] ClassifierCompositeItemProcessor — 분기
    // ======================================================================

    @Bean
    public ItemProcessor<Order, Settlement> classifyingProcessor(JdbcTemplate jdbcTemplate) {

        ItemProcessor<Order, Settlement> standard = new GradeFeeSettlementProcessor(jdbcTemplate);
        ItemProcessor<Order, Settlement> vip = new VipSettlementProcessor(jdbcTemplate);
        ItemProcessor<Order, Settlement> drop = order -> null;   // 정산 대상 아님

        // 분류에 필요한 정보는 "미리 캐시된 맵" 이어야 합니다.
        // classify() 안에서 DB 를 조회하면 [7-3] 에서 없앤 왕복이 되살아납니다.
        Map<Integer, BigDecimal> rates = loadFeeRates(jdbcTemplate);
        BigDecimal vipRate = new BigDecimal("0.0200");

        ClassifierCompositeItemProcessor<Order, Settlement> processor =
                new ClassifierCompositeItemProcessor<>();

        Classifier<Order, ItemProcessor<?, ? extends Settlement>> classifier = order -> {
            if (!"COMPLETED".equals(order.status())) return drop;
            return vipRate.compareTo(rates.get(order.customerId())) == 0 ? vip : standard;
        };
        processor.setClassifier(classifier);
        return processor;
    }

    // ======================================================================
    // [7-9] ValidatingItemProcessor / BeanValidatingItemProcessor
    // ======================================================================

    /**
     * setFilter(false) — 기본값. 위반 시 ValidationException 을 던집니다.
     *   -> "이건 버그다"
     * setFilter(true)  — 위반 시 null 을 반환합니다.
     *   -> "이건 대상이 아니다" (filterCount 로 잡힘)
     */
    @Bean
    public ValidatingItemProcessor<Order> orderValidator() {
        ValidatingItemProcessor<Order> validator = new ValidatingItemProcessor<>(order -> {
            if (order.amount() == null || order.amount().signum() <= 0) {
                throw new ValidationException("금액이 0 이하: order_id=" + order.order_id());
            }
            if (order.orderedAt() == null) {
                throw new ValidationException("주문일시 없음: order_id=" + order.order_id());
            }
        });
        validator.setFilter(false);
        return validator;
    }

    /**
     * Jakarta Bean Validation 기반.
     * ⚠️ @Bean 이 아니면 afterPropertiesSet() 이 호출되지 않아 내부 Validator 가 null 입니다.
     *    new 로 만들어 쓴다면 반드시 손으로 afterPropertiesSet() 을 부르세요.
     */
    @Bean
    public BeanValidatingItemProcessor<ValidatedSettlement> beanValidator() throws Exception {
        BeanValidatingItemProcessor<ValidatedSettlement> p = new BeanValidatingItemProcessor<>();
        p.setFilter(false);
        p.afterPropertiesSet();
        return p;
    }

    /** 검증 애너테이션을 붙인 record. 실제 도메인의 Settlement 에 그대로 옮겨도 됩니다. */
    public record ValidatedSettlement(
            @NotNull Long orderId,
            @NotNull Integer customerId,
            @NotNull BigDecimal grossAmount,
            @NotNull BigDecimal feeRate,
            @NotNull BigDecimal feeAmount,
            @NotNull BigDecimal netAmount) {
    }

    // ======================================================================
    // [7-10] 상태를 가진 프로세서 — 하면 안 되는 예
    // ======================================================================

    /**
     * ❌ 상태를 가진 프로세서.
     *
     * (1) 재시작하면 processed 가 0 부터 시작합니다. 5만 건까지 처리하고 죽으면
     *     재시작 후 최종 리포트가 20,000 이라고 찍힙니다. 에러 없이 틀립니다.
     * (2) .taskExecutor(...) 를 붙이는 순간 여러 스레드가 같은 인스턴스를 씁니다.
     *     processed++ 는 원자적이지 않고, runningTotal 참조 교체도 경쟁 상태입니다.
     *
     * 재현: filteringStep 에 아래 한 줄을 추가하고 여러 번 돌려 보세요.
     *     .taskExecutor(new SimpleAsyncTaskExecutor("batch-"))
     * 매번 다른 값이 나오는 것이 관찰 포인트입니다.
     */
    public static class CountingProcessor implements ItemProcessor<Order, Settlement> {

        private int processed = 0;                            // ← 상태
        private BigDecimal runningTotal = BigDecimal.ZERO;     // ← 상태
        private final ItemProcessor<Order, Settlement> delegate;

        public CountingProcessor(ItemProcessor<Order, Settlement> delegate) {
            this.delegate = delegate;
        }

        @Override
        public Settlement process(Order order) throws Exception {
            processed++;
            runningTotal = runningTotal.add(order.amount());
            return delegate.process(order);
        }

        public int getProcessed() { return processed; }
        public BigDecimal getRunningTotal() { return runningTotal; }
    }

    /**
     * △ AtomicLong 으로 바꾸면 (2) 멀티스레드 문제는 사라집니다.
     *   그러나 (1) 재시작 시 0 부터 세는 문제는 그대로입니다. 정답이 아닙니다.
     *   누적값을 재시작에도 이어가려면 ExecutionContext 에 넣어야 합니다(Step 09).
     */
    public static class AtomicCountingProcessor implements ItemProcessor<Order, Settlement> {

        private final AtomicLong processed = new AtomicLong();
        private final ItemProcessor<Order, Settlement> delegate;

        public AtomicCountingProcessor(ItemProcessor<Order, Settlement> delegate) {
            this.delegate = delegate;
        }

        @Override
        public Settlement process(Order order) throws Exception {
            processed.incrementAndGet();
            return delegate.process(order);
        }
    }

    /**
     * ✅ 정답 — 세지 마세요. 프레임워크가 이미 세고 있습니다([7-4]).
     *   건수: StepExecution.getWriteCount()
     *   누적: ExecutionContext (Step 09)
     *   리포트: StepExecutionListener.afterStep() (Step 12)
     */

    // ======================================================================
    // Reader / Writer / Step / Job
    // ======================================================================

    /** [7-4] status 조건 없이 10만 건 전부 읽습니다. 필터링은 프로세서가 합니다. */
    @Bean
    public ItemReader<Order> allOrdersReader(DataSource dataSource) {
        return new JdbcCursorItemReaderBuilder<Order>()
                .name("allOrdersReader")
                .dataSource(dataSource)
                .sql("SELECT order_id, customer_id, amount, status, ordered_at "
                        + "FROM orders ORDER BY order_id")
                .rowMapper((rs, rowNum) -> new Order(
                        rs.getLong("order_id"),
                        rs.getInt("customer_id"),
                        rs.getBigDecimal("amount"),
                        rs.getString("status"),
                        rs.getTimestamp("ordered_at").toLocalDateTime()))
                .fetchSize(1000)
                .build();
    }

    /** Step 08 에서 자세히 다룹니다. record 라 beanMapped() 대신 람다를 씁니다. */
    @Bean
    public JdbcBatchItemWriter<Settlement> settlementWriter(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Settlement>()
                .dataSource(dataSource)
                .sql("""
                        INSERT INTO settlement
                          (order_id, customer_id, settle_date, gross_amount,
                           fee_rate, fee_amount, net_amount)
                        VALUES
                          (:orderId, :customerId, :settleDate, :grossAmount,
                           :feeRate, :feeAmount, :netAmount)
                        """)
                .itemSqlParameterSourceProvider(s -> new MapSqlParameterSource()
                        .addValue("orderId", s.orderId())
                        .addValue("customerId", s.customerId())
                        .addValue("settleDate", s.settleDate())
                        .addValue("grossAmount", s.grossAmount())
                        .addValue("feeRate", s.feeRate())
                        .addValue("feeAmount", s.feeAmount())
                        .addValue("netAmount", s.netAmount()))
                .assertUpdates(true)
                .build();
    }

    @Bean
    public Step filteringStep(JobRepository jobRepository,
                              PlatformTransactionManager tx,
                              ItemReader<Order> allOrdersReader,
                              ItemProcessor<Order, Settlement> settlementPipeline,
                              ItemWriter<Settlement> settlementWriter,
                              StepExecutionListener countLogger) {
        return new StepBuilder("filteringStep", jobRepository)
                .<Order, Settlement>chunk(1000, tx)
                .reader(allOrdersReader)
                .processor(settlementPipeline)
                .writer(settlementWriter)
                .listener(countLogger)
                // [7-10] 함정 재현용. 평소에는 주석 처리해 둡니다.
                // .taskExecutor(new SimpleAsyncTaskExecutor("batch-"))
                .build();
    }

    @Bean
    public Step classifyingStep(JobRepository jobRepository,
                                PlatformTransactionManager tx,
                                ItemReader<Order> allOrdersReader,
                                ItemProcessor<Order, Settlement> classifyingProcessor,
                                ItemWriter<Settlement> settlementWriter,
                                StepExecutionListener countLogger) {
        return new StepBuilder("classifyingStep", jobRepository)
                .<Order, Settlement>chunk(1000, tx)
                .reader(allOrdersReader)
                .processor(classifyingProcessor)
                .writer(settlementWriter)
                .listener(countLogger)
                .build();
    }

    @Bean
    public Job filteringJob(JobRepository jobRepository, Step filteringStep) {
        return new JobBuilder("filteringJob", jobRepository)
                .start(filteringStep)
                .build();
    }

    @Bean
    public Job classifyingJob(JobRepository jobRepository, Step classifyingStep) {
        return new JobBuilder("classifyingJob", jobRepository)
                .start(classifyingStep)
                .build();
    }

    /** SimpleAsyncTaskExecutor import 를 실제로 쓰는 자리(멀티스레드 재현용). */
    static SimpleAsyncTaskExecutor batchExecutor() {
        SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("batch-");
        executor.setConcurrencyLimit(4);
        return executor;
    }
}

package com.example.batch.step07;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.support.ClassifierCompositeItemProcessor;
import org.springframework.batch.item.support.CompositeItemProcessor;
import org.springframework.jdbc.core.JdbcTemplate;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Step 07 — 연습문제 6문항.
 *
 * 규칙
 *  - "여기에 작성:" 자리를 채우세요.
 *  - 문제 3·4 는 코드보다 "예측"이 본체입니다. 실행하기 전에 반드시 주석에 예측을 적으세요.
 *  - 실행 전마다 초기화:
 *      mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 *
 * 정답은 Solution.java.
 */
public class Exercise {

    // 공용 헬퍼 — customers 1,000행의 fee_rate 맵
    static Map<Integer, BigDecimal> loadFeeRates(JdbcTemplate jdbcTemplate) {
        Map<Integer, BigDecimal> map = new HashMap<>(1400);
        jdbcTemplate.query("SELECT customer_id, fee_rate FROM customers",
                rs -> { map.put(rs.getInt("customer_id"), rs.getBigDecimal("fee_rate")); });
        return Map.copyOf(map);
    }

    // ======================================================================
    // 문제 1 — Order -> Settlement 변환과 반올림 규칙
    // ======================================================================
    /**
     * 요구사항
     *  - 고객 등급별 fee_rate 를 적용해 Settlement 을 만든다.
     *  - settle_date 는 ordered_at 의 날짜 부분.
     *  - 실행 후 다음이 성립해야 한다:
     *      SELECT SUM(gross_amount) - SUM(fee_amount) = SUM(net_amount) FROM settlement;
     *
     * 힌트: BigDecimal.multiply 는 두 피연산자의 스케일을 더합니다(2 + 4 = 6).
     *       어느 시점에 setScale(2, HALF_UP) 을 해야 위 등식이 성립할까요?
     */
    public static class Q1_SettlementProcessor implements ItemProcessor<Order, Settlement> {

        private final Map<Integer, BigDecimal> feeRateByCustomer;

        public Q1_SettlementProcessor(JdbcTemplate jdbcTemplate) {
            this.feeRateByCustomer = loadFeeRates(jdbcTemplate);
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal rate = feeRateByCustomer.get(order.customerId());
            BigDecimal gross = order.amount();

            // 여기에 작성: fee 와 net 을 계산하고 Settlement 을 반환하세요.
            //             반올림을 "어디서 한 번" 할지가 핵심입니다.

            return null;
        }
    }

    // ======================================================================
    // 문제 2 — null 필터링과 filterCount
    // ======================================================================
    /**
     * 리더는 WHERE 절 없이 orders 10만 건을 전부 읽습니다(이 조건을 되돌리지 마세요).
     * status 가 COMPLETED 가 아닌 아이템을 걸러 내세요.
     *
     * 실행 후 확인:
     *   SELECT READ_COUNT, WRITE_COUNT, FILTER_COUNT
     *   FROM BATCH_STEP_EXECUTION ORDER BY STEP_EXECUTION_ID DESC LIMIT 1;
     *
     * 기대값을 먼저 적어 두세요.
     *   readCount   = ______
     *   writeCount  = ______
     *   filterCount = ______
     */
    public static class Q2_CompletedOnly implements ItemProcessor<Order, Order> {

        @Override
        public Order process(Order order) {
            // 여기에 작성:

            return order;
        }
    }

    // ======================================================================
    // 문제 3 — 예외 방식으로 바꾸면 통계가 어떻게 달라지는가 (예측 문제)
    // ======================================================================
    /**
     * 아래는 문제 2 와 "같은 비즈니스 규칙"을 예외로 구현한 것입니다.
     * Step 설정에는 .faultTolerant().skip(IllegalArgumentException.class).skipLimit(50000)
     * 이 붙어 있다고 가정합니다.
     *
     * 실행하기 전에 예측을 적으세요.
     *   filterCount        = ______
     *   processSkipCount   = ______
     *   rollbackCount      = 대략 ______
     *   Job 최종 STATUS    = ______
     *
     * 그리고 "왜 rollbackCount 가 그 값인가"를 한 문장으로 적으세요.
     *   여기에 작성:
     *
     * 마지막으로: 이 구현이 왜 나쁜지 세 가지를 적으세요.
     *   (1) 여기에 작성:
     *   (2) 여기에 작성:
     *   (3) 여기에 작성:
     */
    public static class Q3_ExceptionBased implements ItemProcessor<Order, Order> {

        @Override
        public Order process(Order order) {
            if (!"COMPLETED".equals(order.status())) {
                throw new IllegalArgumentException("정산 대상 아님: " + order.order_id());
            }
            return order;
        }
    }

    // ======================================================================
    // 문제 4 — CompositeItemProcessor 의 타입 함정 (예측 + 리팩터링)
    // ======================================================================
    /**
     * (a) 아래 파이프라인은 컴파일됩니다. 실행하면 어떤 일이 벌어질까요?
     *     예외 클래스 이름 = ______________________
     *     스택트레이스가 가리키는 클래스 = ______________________
     *     실제로 고쳐야 하는 코드의 위치 = ______________________
     *
     * (b) 그다음, 같은 파이프라인을 "순서를 틀리면 컴파일이 실패하도록" 다시 쓰세요.
     */
    public static class Q4_BrokenComposite {

        public ItemProcessor<Order, Settlement> broken(JdbcTemplate jdbcTemplate) {
            CompositeItemProcessor<Order, Settlement> composite = new CompositeItemProcessor<>();
            composite.setDelegates(List.of(
                    new Q1_SettlementProcessor(jdbcTemplate),   // Order -> Settlement
                    new Q2_CompletedOnly()                      // Order 를 기대한다
            ));
            return composite;
        }

        /** (b) 컴파일러가 순서를 검사해 주는 형태로 다시 작성하세요. */
        public ItemProcessor<Order, Settlement> fixed(JdbcTemplate jdbcTemplate) {
            // 여기에 작성:

            return null;
        }
    }

    // ======================================================================
    // 문제 5 — ClassifierCompositeItemProcessor 분기
    // ======================================================================
    /**
     * 규칙
     *  - status != 'COMPLETED'  -> null (정산 제외)
     *  - fee_rate == 0.0200 (VIP) -> 수수료율에서 0.0050 을 추가 할인
     *  - 그 외                   -> 표준 처리
     *
     * 제약: classify() 안에서 DB 를 조회하면 안 됩니다. 이유도 주석으로 적으세요.
     *   여기에 작성(이유):
     *
     * 실행 후 확인:
     *   SELECT c.grade, s.fee_rate, COUNT(*) FROM settlement s
     *   JOIN customers c ON c.customer_id = s.customer_id
     *   GROUP BY c.grade, s.fee_rate ORDER BY s.fee_rate DESC;
     * VIP 행의 fee_rate 가 ______ 이어야 합니다.
     */
    public static class Q5_Classifying {

        public ItemProcessor<Order, Settlement> build(JdbcTemplate jdbcTemplate) {
            ClassifierCompositeItemProcessor<Order, Settlement> processor =
                    new ClassifierCompositeItemProcessor<>();

            // 여기에 작성: standard / vip / drop 프로세서를 만들고 classifier 를 설정하세요.

            return processor;
        }
    }

    // ======================================================================
    // 문제 6 — 상태를 가진 프로세서 리팩터링
    // ======================================================================
    /**
     * 아래 프로세서는 단일 스레드로 돌리면 정확한 값을 냅니다.
     * Job 이 COMPLETED 로 끝나도 통과한 것이 아닙니다.
     *
     * (a) 이 코드가 틀리는 두 상황을 적으세요.
     *     상황 1: 여기에 작성:
     *     상황 2: 여기에 작성:
     *
     * (b) processed 를 AtomicLong 으로 바꾸면 두 상황이 모두 해결됩니까? 답과 이유:
     *     여기에 작성:
     *
     * (c) 무상태로 리팩터링하세요. 처리 건수와 누적 합계를 각각 어디서 얻어야 합니까?
     *     처리 건수: 여기에 작성:
     *     누적 합계: 여기에 작성:
     */
    public static class Q6_Stateful implements ItemProcessor<Order, Settlement> {

        private int processed = 0;
        private BigDecimal runningTotal = BigDecimal.ZERO;
        private final ItemProcessor<Order, Settlement> delegate;

        public Q6_Stateful(ItemProcessor<Order, Settlement> delegate) {
            this.delegate = delegate;
        }

        @Override
        public Settlement process(Order order) throws Exception {
            processed++;
            runningTotal = runningTotal.add(order.amount());
            return delegate.process(order);
        }
    }

    /** (c) 여기에 무상태 버전을 작성하세요. */
    public static class Q6_Stateless implements ItemProcessor<Order, Settlement> {

        // 여기에 작성: 필드는 전부 final + 불변이어야 합니다.

        @Override
        public Settlement process(Order order) {
            // 여기에 작성:

            return null;
        }
    }
}

package com.example.batch.step07;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.support.ClassifierCompositeItemProcessor;
import org.springframework.classify.Classifier;
import org.springframework.jdbc.core.JdbcTemplate;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.Map;

/**
 * Step 07 — 연습문제 정답과 해설.
 *
 * 문제를 직접 풀어 본 뒤에 여세요.
 */
public class Solution {

    static Map<Integer, BigDecimal> loadFeeRates(JdbcTemplate jdbcTemplate) {
        Map<Integer, BigDecimal> map = new HashMap<>(1400);
        jdbcTemplate.query("SELECT customer_id, fee_rate FROM customers",
                rs -> { map.put(rs.getInt("customer_id"), rs.getBigDecimal("fee_rate")); });
        return Map.copyOf(map);   // 불변화 — 멀티스레드에서도 안전
    }

    // ======================================================================
    // 정답 1 — 반올림은 "한 번", "가장 이른 시점"에
    // ======================================================================
    /*
     * 핵심은 계산 순서입니다.
     *
     *   ✅ fee = gross.multiply(rate).setScale(2, HALF_UP);   // 여기서 2자리로 "확정"
     *      net = gross.subtract(fee);                          // 확정된 fee 로 뺀다
     *
     *   ❌ fee = gross.multiply(rate);                         // 스케일 6
     *      net = gross.subtract(fee);                          // net 도 스케일 6
     *      // DB 가 행마다 알아서 2자리로 반올림한다
     *
     * 왜 ❌ 가 문제인가:
     *   gross=1050.00, rate=0.0250 이면 fee = 26.250000, net = 1023.750000 입니다.
     *   DB 의 DECIMAL(12,2) 는 각 행을 독립적으로 반올림하므로
     *     fee -> 26.25 (그대로), net -> 1023.75 (그대로)  ← 이 경우는 우연히 맞습니다.
     *   그러나 gross=1025.00, rate=0.0350 이면 fee = 35.875000, net = 989.125000 이고
     *     fee -> 35.88 (올림), net -> 989.13 (올림)
     *   35.88 + 989.13 = 1025.01 != 1025.00.  한 건당 0.01원이 "생겨납니다".
     *
     *   ✅ 방식이면 fee = 35.88 로 먼저 확정하고 net = 1025.00 - 35.88 = 989.12 이므로
     *   언제나 gross = fee + net 이 성립합니다.
     *
     * 7만 건 기준 실제 검증 결과:
     *   +-------+---------------+--------------+---------------+
     *   | cnt   | gross         | fee          | net           |
     *   | 70000 | 3485250000.00 |  97089107.00 | 3388160893.00 |
     *   +-------+---------------+--------------+---------------+
     *   3485250000.00 - 97089107.00 = 3388160893.00  ← 정확히 일치
     *
     * 교훈: 반올림은 "회계 규칙"이지 "표시 형식"이 아닙니다.
     *       어느 시점에 몇 자리로 확정할지를 코드가 명시적으로 정해야 합니다.
     *       DB 에 맡기면, DB 는 각 컬럼을 서로 모른 채 따로 반올림합니다.
     */
    public static class A1_SettlementProcessor implements ItemProcessor<Order, Settlement> {

        private final Map<Integer, BigDecimal> feeRateByCustomer;

        public A1_SettlementProcessor(JdbcTemplate jdbcTemplate) {
            this.feeRateByCustomer = loadFeeRates(jdbcTemplate);
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal rate = feeRateByCustomer.get(order.customerId());
            if (rate == null) {
                throw new IllegalStateException("등급 정보 없는 고객: " + order.customerId());
            }
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            BigDecimal net = gross.subtract(fee);

            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(), gross, rate, fee, net);
        }
    }

    // ======================================================================
    // 정답 2 — null 반환. 기대값 readCount=100000, writeCount=70000, filterCount=30000
    // ======================================================================
    /*
     * orders 는 100,000행이고 status 분포는 COMPLETED 70,000 / CANCELLED 10,000 /
     * PENDING 10,000 / REFUNDED 10,000 입니다. 따라서 30,000건이 걸러집니다.
     *
     * 실제 BATCH_STEP_EXECUTION:
     *   READ_COUNT=100000, WRITE_COUNT=70000, FILTER_COUNT=30000,
     *   COMMIT_COUNT=101, ROLLBACK_COUNT=0, PROCESS_SKIP_COUNT=0
     *
     * COMMIT_COUNT 가 101 인 이유: 100,000 / 1,000 = 100 청크 + "더 읽을 게 없음"을
     * 확인하는 마지막 빈 커밋 1회.
     *
     * 항등식: readCount = writeCount + filterCount + skipCount
     *         100000    = 70000      + 30000       + 0
     */
    public static class A2_CompletedOnly implements ItemProcessor<Order, Order> {

        @Override
        public Order process(Order order) {
            if (!"COMPLETED".equals(order.status())) {
                return null;   // 예외가 아닙니다. 정상 흐름입니다.
            }
            return order;
        }
    }

    // ======================================================================
    // 정답 3 — 예외 방식으로 바꾸면
    // ======================================================================
    /*
     * 예측 정답:
     *   filterCount      = 0
     *   processSkipCount = 30000
     *   rollbackCount    ≈ 30000 (스킵 건수와 거의 같음)
     *   Job STATUS       = COMPLETED   (skipLimit=50000 이라 한도 안에 들어옴)
     *                      skipLimit 을 10 으로 두었다면 FAILED 입니다.
     *
     * rollbackCount 가 왜 그렇게 큰가:
     *   Spring Batch 의 스킵은 "예외가 난 아이템만 빼고 계속"이 아닙니다.
     *   process 에서 스킵 가능한 예외가 나면 프레임워크는
     *     (1) 청크 트랜잭션 전체를 롤백하고
     *     (2) 그 청크를 "아이템 하나씩 커밋하는 모드"로 재시도해
     *     (3) 문제 아이템만 골라내 스킵한 뒤 나머지를 다시 처리합니다.
     *   즉 스킵 1건이 청크 하나의 재실행을 부릅니다.
     *   30,000건이 흩어져 있으므로 거의 모든 청크가 최소 한 번씩 롤백됩니다.
     *
     * 이 구현이 나쁜 세 가지:
     *   (1) 통계가 오염된다.
     *       "정상적으로 제외한 30,000건"과 "장애로 버린 건"이 같은 칸에 섞입니다.
     *       모니터링의 skipCount 알람이 영구히 무의미해집니다.
     *   (2) skipLimit 을 넘기면 Job 이 FAILED 로 끝난다.
     *       데이터는 아무 문제가 없는데 배치가 실패합니다.
     *       그리고 skipLimit 을 데이터 분포에 맞춰 계속 키워야 하는 악순환이 생깁니다.
     *   (3) 비싸다.
     *       예외 30,000개 생성 + 스택트레이스 채우기 + 청크 롤백/재처리.
     *       측정하면 null 방식 7s034ms 대비 40초 이상으로 벌어집니다.
     *
     * 원칙 한 줄:
     *   "정상적으로 제외할 것은 null 로, 진짜 잘못된 데이터만 예외로."
     */

    // ======================================================================
    // 정답 4 — 타입 함정과 컴파일 타임 검증
    // ======================================================================
    /*
     * (a)
     *   예외 클래스        : java.lang.ClassCastException
     *                        "class ...domain.Settlement cannot be cast to class ...domain.Order"
     *   스택트레이스가 가리키는 곳 : Q2_CompletedOnly.process()  (2번째 델리게이트)
     *   실제로 고쳐야 하는 곳      : composite.setDelegates(List.of(...)) 의 "순서"
     *
     *   이 어긋남이 이 함정의 본질입니다. 터지는 위치와 원인 위치가 다릅니다.
     *   처음 보면 Q2_CompletedOnly 를 열어 놓고 한참을 들여다보게 됩니다.
     *
     *   왜 컴파일러가 못 잡는가:
     *     public void setDelegates(List<? extends ItemProcessor<?, ?>> delegates)
     *   파라미터 타입이 ItemProcessor<?, ?> 입니다. 어떤 프로세서든 들어갑니다.
     *   CompositeItemProcessor.process() 안에서는
     *     result = processItem((ItemProcessor<Object, Object>) delegate, result);
     *   로 캐스팅해 돌리므로, 실제 검사는 델리게이트 메서드 진입 시점의
     *   체크캐스트에서야 일어납니다.
     *
     *   더 위험한 변형: 두 타입이 캐스팅 가능한 관계(상속 관계, 혹은 둘 다 Map)라면
     *   예외조차 나지 않고 엉뚱한 필드를 읽어 조용히 틀린 값을 씁니다.
     *
     * (b) 아래 fixed() 처럼 직접 합성하면 컴파일러가 검사합니다.
     *     순서를 바꿔 step3 을 먼저 호출하도록 고쳐 보세요:
     *       Order a = step3.process(order);
     *     ->  error: incompatible types: Settlement cannot be converted to Order
     *     빌드 단계에서 즉시 잡힙니다.
     *
     *   대가: null 전파를 손으로 써야 합니다. 단계가 3~4개로 고정된 파이프라인이라면
     *   이 대가가 훨씬 쌉니다. 델리게이트를 런타임에 조립해야 하는 경우에만
     *   CompositeItemProcessor 를 쓰고, 그때는 반드시 아이템 1건짜리 단위 테스트를
     *   붙이세요. composite.process(sampleOrder) 한 줄이면 잡힙니다.
     */
    public static class A4_Fixed {

        public ItemProcessor<Order, Settlement> fixed(JdbcTemplate jdbcTemplate) {
            ItemProcessor<Order, Order> step1 = new A2_CompletedOnly();
            ItemProcessor<Order, Settlement> step2 = new A1_SettlementProcessor(jdbcTemplate);

            return order -> {
                Order filtered = step1.process(order);
                if (filtered == null) return null;      // null 전파를 명시적으로
                return step2.process(filtered);
            };
        }
    }

    // ======================================================================
    // 정답 5 — ClassifierCompositeItemProcessor
    // ======================================================================
    /*
     * classify() 안에서 DB 를 조회하면 안 되는 이유:
     *   classify() 는 process() 와 마찬가지로 "아이템마다 한 번" 호출됩니다.
     *   여기서 SELECT 를 날리면 7-3 에서 캐시로 없앤 7만 번의 왕복이 그대로 되살아납니다.
     *   분류는 "이미 손에 든 정보"만으로 끝나야 합니다.
     *   필요한 정보는 (a) 리더가 실어 온 필드이거나 (b) 생성 시점에 캐시된 맵이어야 합니다.
     *
     * VIP 행의 fee_rate 는 0.0150 (= 0.0200 - 0.0050) 입니다.
     *
     * 실행 결과:
     *   +--------+----------+-------+
     *   | grade  | fee_rate | cnt   |
     *   +--------+----------+-------+
     *   | BRONZE |   0.0350 | 20000 |
     *   | SILVER |   0.0300 | 15000 |
     *   | GOLD   |   0.0250 | 20000 |
     *   | VIP    |   0.0150 | 15000 |
     *   +--------+----------+-------+
     *   readCount=100000, writeCount=70000, filterCount=30000, skipCount=0
     *
     * 참고: 아이템의 "클래스 타입"으로만 분기하면 되는 흔한 경우에는
     *   new SubclassClassifier<>(Map.of(A.class, p1, B.class, p2), defaultP)
     * 한 줄로 끝납니다.
     */
    public static class A5_Classifying {

        private static final BigDecimal VIP_RATE = new BigDecimal("0.0200");
        private static final BigDecimal VIP_DISCOUNT = new BigDecimal("0.0050");

        public ItemProcessor<Order, Settlement> build(JdbcTemplate jdbcTemplate) {

            Map<Integer, BigDecimal> rates = loadFeeRates(jdbcTemplate);

            ItemProcessor<Order, Settlement> standard = new A1_SettlementProcessor(jdbcTemplate);

            ItemProcessor<Order, Settlement> vip = order -> {
                BigDecimal rate = rates.get(order.customerId()).subtract(VIP_DISCOUNT);
                BigDecimal gross = order.amount();
                BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
                return new Settlement(order.order_id(), order.customerId(),
                        order.orderedAt().toLocalDate(), gross, rate, fee, gross.subtract(fee));
            };

            ItemProcessor<Order, Settlement> drop = order -> null;

            ClassifierCompositeItemProcessor<Order, Settlement> processor =
                    new ClassifierCompositeItemProcessor<>();

            Classifier<Order, ItemProcessor<?, ? extends Settlement>> classifier = order -> {
                if (!"COMPLETED".equals(order.status())) return drop;   // filterCount 로 잡힘
                return VIP_RATE.compareTo(rates.get(order.customerId())) == 0 ? vip : standard;
            };
            processor.setClassifier(classifier);
            return processor;
        }
    }

    // ======================================================================
    // 정답 6 — 무상태 리팩터링
    // ======================================================================
    /*
     * (a) 틀리는 두 상황
     *   상황 1: 재시작.
     *     7만 건 중 5만 건까지 처리하고 죽은 뒤 재시작하면, 리더는 ExecutionContext 덕에
     *     5만 건째부터 이어가지만 processed 필드는 새 인스턴스라 0 입니다.
     *     최종 리포트가 20,000 으로 찍힙니다. 예외도 경고도 없습니다.
     *   상황 2: 멀티스레드.
     *     .taskExecutor(...) 를 붙이면 하나의 프로세서 인스턴스를 여러 스레드가 공유합니다.
     *     processed++ 는 읽고-더하고-쓰는 세 동작이라 원자적이지 않고,
     *     runningTotal = runningTotal.add(...) 도 참조 교체라 덮어쓰기가 일어납니다.
     *     BigDecimal 이 불변인 것은 도움이 되지 않습니다 — 불변인 것은 "값"이지
     *     "필드 참조"가 아니기 때문입니다.
     *     실측: processed=68847, runningTotal=3421880400.00 (기대 70000 / 3485250000.00)
     *
     * (b) AtomicLong 으로 바꾸면?
     *     아니오. 상황 2 만 해결되고 상황 1 은 그대로입니다.
     *     AtomicLong 도 JVM 힙에 있는 값이라 프로세스가 죽으면 사라집니다.
     *     "동시성 문제"와 "영속성 문제"는 다른 문제입니다.
     *     이 오답이 흔한 이유는, 멀티스레드에서 틀린 값을 보고 원인을 동시성 하나로만
     *     좁혀 버리기 때문입니다.
     *
     * (c) 어디서 얻어야 하는가
     *     처리 건수 : StepExecution.getWriteCount() / getFilterCount() / getReadCount().
     *                 프레임워크가 이미 세고 있고, BATCH_STEP_EXECUTION 에 영속되므로
     *                 재시작에도 살아남습니다. 직접 셀 이유가 없습니다.
     *     누적 합계 : StepExecution 의 ExecutionContext.
     *                 청크 커밋마다 함께 커밋되므로 재시작 시 마지막 커밋 지점의 값이
     *                 복원됩니다. (Step 09 에서 다룹니다)
     *                 리스너에서 @BeforeStep 으로 StepExecution 을 주입받아
     *                 executionContext.put("runningTotal", ...) 로 갱신합니다.
     *     최종 리포트: StepExecutionListener.afterStep() 또는 별도 Tasklet Step (Step 12).
     *
     * 규칙 한 줄:
     *   ItemProcessor 구현체의 모든 인스턴스 필드에 final 을 붙이고,
     *   그 안에 담기는 컬렉션도 불변으로 만든다.
     *   final 을 못 붙이겠다는 필드가 생기면, 그게 설계가 잘못됐다는 신호입니다.
     */
    public static class A6_Stateless implements ItemProcessor<Order, Settlement> {

        private final Map<Integer, BigDecimal> feeRateByCustomer;   // final + 불변

        public A6_Stateless(JdbcTemplate jdbcTemplate) {
            this.feeRateByCustomer = loadFeeRates(jdbcTemplate);
        }

        @Override
        public Settlement process(Order order) {
            BigDecimal rate = feeRateByCustomer.get(order.customerId());
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(), gross, rate, fee, gross.subtract(fee));
        }
    }
}