#Step 05 — 청크 지향 처리

// 이렇게 하면 안 됩니다
public RepeatStatus execute(StepContribution c, ChunkContext ctx) {
    List<Order> all = jdbc.query("SELECT * FROM orders WHERE status='COMPLETED'", mapper);
    List<Settlement> results = all.stream().map(this::settle).toList();
    jdbc.batchUpdate("INSERT INTO settlement ...", results);
    return RepeatStatus.FINISHED;
}

@Bean
public Step settlementStep(JobRepository jobRepository,
                           PlatformTransactionManager txManager,
                           ItemReader<Order> orderReader,
                           ItemProcessor<Order, Settlement> settlementProcessor,
                           ItemWriter<Settlement> settlementWriter) {
    return new StepBuilder("settlementStep", jobRepository)
            .<Order, Settlement>chunk(1000, txManager)   // ← 5.x: 트랜잭션 매니저가 인자
            .reader(orderReader)
            .processor(settlementProcessor)
            .writer(settlementWriter)
            .build();
}

// "1,000건을 모으거나, 3초가 지나면 커밋" — 둘 중 먼저 오는 쪽
.<Order, Settlement>chunk(new SimpleCompletionPolicy(1000), txManager)

// 청크 크기를 고정하지 않고 정책으로 위임
.<Order, Settlement>chunk(new TimeoutTerminationPolicy(3000), txManager)

// [5-7] 25,501번째 아이템에서 터지는 processor
static class ExplodingProcessor implements ItemProcessor<Order, Settlement> {

    private final AtomicInteger seq = new AtomicInteger();
    private final SettlementProcessor delegate;

    ExplodingProcessor(SettlementProcessor delegate) {
        this.delegate = delegate;
    }

    @Override
    public Settlement process(Order item) throws Exception {
        int n = seq.incrementAndGet();
        if (n == 25_501) {
            throw new IllegalStateException("의도적 실패: " + n + "번째 아이템 (order_id=" + item.order_id() + ")");
        }
        return delegate.process(item);
    }
}

// 4.x
public interface ItemWriter<T> {
    void write(List<? extends T> items) throws Exception;
}

// 5.x
public interface ItemWriter<T> {
    void write(Chunk<? extends T> chunk) throws Exception;
}

// [5-8] 5.x writer 구현
static class SettlementLogWriter implements ItemWriter<Settlement> {

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

    @Override
    public void write(Chunk<? extends Settlement> chunk) {
        // Chunk 는 Iterable 이므로 향상된 for 문이 그대로 됩니다
        BigDecimal sum = BigDecimal.ZERO;
        for (Settlement s : chunk) {
            sum = sum.add(s.netAmount());
        }
        // List 가 필요하면 getItems()
        List<? extends Settlement> items = chunk.getItems();
        log.debug("청크 {}건, 첫 주문 {}, 순액 합계 {}",
                chunk.size(), items.get(0).orderId(), sum);
    }
}

// [5-8] 테스트에서 Chunk 만들기
@Test
void writerInsertsAllRows() throws Exception {
    Chunk<Settlement> chunk = Chunk.of(
            new Settlement(1L, 1, LocalDate.of(2025, 1, 1),
                    new BigDecimal("1000.00"), new BigDecimal("0.0300"),
                    new BigDecimal("30.00"), new BigDecimal("970.00")),
            new Settlement(2L, 2, LocalDate.of(2025, 1, 1),
                    new BigDecimal("2000.00"), new BigDecimal("0.0250"),
                    new BigDecimal("50.00"), new BigDecimal("1950.00"))
    );

    writer.write(chunk);

    assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM settlement", Integer.class)).isEqualTo(2);
}

package com.example.batch.step05;

/*
 * ============================================================================
 * Step 05 — 청크 지향 처리 / 실습 코드
 * ============================================================================
 *
 * 본문 5-2 ~ 5-9 의 모든 예제를 절 번호 주석과 함께 담았습니다.
 *
 * [실행 전 초기화]
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb <<'SQL'
 *   SET FOREIGN_KEY_CHECKS = 0;
 *   DELETE FROM BATCH_STEP_EXECUTION_CONTEXT;
 *   DELETE FROM BATCH_STEP_EXECUTION;
 *   DELETE FROM BATCH_JOB_EXECUTION_CONTEXT;
 *   DELETE FROM BATCH_JOB_EXECUTION_PARAMS;
 *   DELETE FROM BATCH_JOB_EXECUTION;
 *   DELETE FROM BATCH_JOB_INSTANCE;
 *   SET FOREIGN_KEY_CHECKS = 1;
 *   TRUNCATE TABLE settlement;
 *   SQL
 *
 * [실행]
 *   ./gradlew bootRun --args='--spring.batch.job.name=settlementChunkJob --batch.chunk-size=1000'
 *
 *   ⚠️ --spring.batch.job.name 을 반드시 지정하세요.
 *      지정하지 않으면 Boot 3.2 가 이 파일의 Job 을 전부 순차 실행합니다.
 *
 * [5-6 실측]
 *   for SIZE in 10 100 1000 10000; do
 *     mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 *     ./gradlew bootRun --args="--spring.batch.job.name=settlementChunkJob --batch.chunk-size=$SIZE"
 *   done
 * ============================================================================
 */

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
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.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
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.repeat.policy.SimpleCompletionPolicy;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.DataClassRowMapper;
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.time.LocalDate;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class Practice {

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

    // ========================================================================
    // 도메인 — com.example.batch.domain 의 record 를 그대로 옮겨 놓은 것입니다.
    // 실제 프로젝트에서는 domain 패키지의 것을 import 해서 쓰세요.
    // ========================================================================

    public record Order(
            Long order_id,
            Integer customerId,
            BigDecimal amount,
            String status,
            LocalDateTime orderedAt
    ) {}

    public record Settlement(
            Long orderId,
            Integer customerId,
            LocalDate settleDate,
            BigDecimal grossAmount,
            BigDecimal feeRate,
            BigDecimal feeAmount,
            BigDecimal netAmount
    ) {}

    // ========================================================================
    // [5-3] 청크 Step 정의 — Spring Batch 5.1.1 시그니처
    // ========================================================================

    /**
     * 청크 크기는 하드코딩하지 않고 설정으로 뺍니다 (본문 5-9 의 실무 팁).
     * --batch.chunk-size=100 처럼 커맨드라인으로 바꿀 수 있습니다.
     */
    @Value("${batch.chunk-size:1000}")
    private int chunkSize;

    /** 5-5 의 트랜잭션 경계 관찰용. --slow=true 일 때만 켜집니다. */
    @Value("${slow:false}")
    private boolean slow;

    @Bean
    public Step settlementStep(JobRepository jobRepository,
                               PlatformTransactionManager txManager,
                               ItemReader<Order> orderReader,
                               ItemProcessor<Order, Settlement> settlementProcessor,
                               ItemWriter<Settlement> settlementWriter) {

        log.info("settlementStep 생성 — chunkSize={}, slow={}", chunkSize, slow);

        return new StepBuilder("settlementStep", jobRepository)
                // ↓ 5.x 의 핵심. 트랜잭션 매니저가 인자입니다.
                //   4.x 의 .chunk(1000) 만 쓰면 런타임에
                //   IllegalStateException: A transaction manager must be provided
                .<Order, Settlement>chunk(chunkSize, txManager)
                .reader(orderReader)
                .processor(settlementProcessor)
                .writer(settlementWriter)
                .build();
    }

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

    /**
     * [5-3] CompletionPolicy 오버로드 — 참고용.
     * "1,000건을 모으거나 정책이 끝났다고 하면 커밋".
     * 아이템 크기가 들쭉날쭉해서 건수로 메모리를 예측할 수 없을 때 씁니다.
     */
    @Bean
    public Step settlementPolicyStep(JobRepository jobRepository,
                                     PlatformTransactionManager txManager,
                                     ItemReader<Order> orderReader,
                                     ItemProcessor<Order, Settlement> settlementProcessor,
                                     ItemWriter<Settlement> settlementWriter) {
        return new StepBuilder("settlementPolicyStep", jobRepository)
                .<Order, Settlement>chunk(new SimpleCompletionPolicy(chunkSize), txManager)
                .reader(orderReader)
                .processor(settlementProcessor)
                .writer(settlementWriter)
                .build();
    }

    // ========================================================================
    // [5-4] Reader — JdbcCursorItemReader
    //   fetchSize 를 청크 크기와 같게 맞춥니다 (5-9 의 팁).
    //   커서 리더의 한계(커넥션 점유·멀티스레드 불가)는 Step 06 에서 다룹니다.
    // ========================================================================

    @Bean
    public JdbcCursorItemReader<Order> orderReader(DataSource dataSource) {
        return new JdbcCursorItemReaderBuilder<Order>()
                .name("orderReader")
                .dataSource(dataSource)
                .sql("""
                     SELECT order_id, customer_id, amount, status, ordered_at
                     FROM orders
                     WHERE status = 'COMPLETED'
                     ORDER BY order_id
                     """)
                .fetchSize(chunkSize)
                // record 는 자바빈이 아니므로 BeanPropertyRowMapper 가 아니라
                // DataClassRowMapper 를 씁니다. 자세한 이유는 Step 06 에서.
                .rowMapper(new DataClassRowMapper<>(Order.class))
                .build();
    }

    // ========================================================================
    // [5-4] Processor — 수수료율 캐시를 써서 아이템당 DB 조회를 없앱니다.
    //
    //   이 캐시가 없으면 아이템마다 SELECT 가 나가고, 70,000번의 왕복이
    //   전체 시간을 지배해 5-6 의 U자 곡선이 아예 안 보입니다.
    // ========================================================================

    @Bean
    public Map<Integer, BigDecimal> feeRateCache(JdbcTemplate jdbc) {
        Map<Integer, BigDecimal> cache = new HashMap<>();
        jdbc.query("SELECT customer_id, fee_rate FROM customers", rs -> {
            cache.put(rs.getInt("customer_id"), rs.getBigDecimal("fee_rate"));
        });
        log.info("수수료율 캐시 로드 완료 — {}건", cache.size());   // 1000건
        return cache;
    }

    static class SettlementProcessor implements ItemProcessor<Order, Settlement> {

        private final Map<Integer, BigDecimal> feeRates;

        SettlementProcessor(Map<Integer, BigDecimal> feeRates) {
            this.feeRates = feeRates;
        }

        @Override
        public Settlement process(Order item) {
            // ⚠️ 여기서 "이번 청크의 합계" 같은 걸 필드에 누적하면 안 됩니다.
            //    processor 는 아이템 하나만 봅니다 (본문 5-2 의 함정).
            BigDecimal rate = feeRates.get(item.customerId());
            BigDecimal gross = item.amount();
            BigDecimal fee = gross.multiply(rate).setScale(2, RoundingMode.HALF_UP);
            BigDecimal net = gross.subtract(fee);

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

    @Bean
    public ItemProcessor<Order, Settlement> settlementProcessor(Map<Integer, BigDecimal> feeRateCache) {
        return new SettlementProcessor(feeRateCache);
    }

    // ========================================================================
    // [5-4] Writer — JdbcBatchItemWriter
    //
    //   ⚠️ record 에는 BeanPropertyItemSqlParameterSourceProvider 가 안 됩니다.
    //      (record 의 접근자는 getOrderId() 가 아니라 orderId() 라서)
    //      그래서 람다로 MapSqlParameterSource 를 직접 만듭니다.
    //      Step 08 에서 정면으로 다룹니다.
    // ========================================================================

    @Bean
    public JdbcBatchItemWriter<Settlement> settlementWriter(DataSource dataSource) {
        JdbcBatchItemWriter<Settlement> writer = 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()))
                .build();
        writer.afterPropertiesSet();

        if (!slow) {
            return writer;
        }
        // [5-5] --slow=true 일 때만: 청크당 200ms 쉬어서 트랜잭션 경계를
        //       다른 터미널에서 관찰할 수 있게 합니다.
        //       ⚠️ 이 플래그를 켠 채로 5-6 의 실측을 하면 안 됩니다.
        return slowWriter(writer);
    }

    /** [5-5] 트랜잭션 경계 관찰용 래퍼. */
    static JdbcBatchItemWriter<Settlement> slowWriter(JdbcBatchItemWriter<Settlement> delegate) {
        return new JdbcBatchItemWriter<>() {
            @Override
            public void write(Chunk<? extends Settlement> chunk) throws Exception {
                delegate.write(chunk);
                Thread.sleep(200);   // 커밋 직전에 붙잡아 둡니다
            }
        };
    }

    // ========================================================================
    // [5-7] 롤백 단위가 청크임을 확인하는 Job
    //
    //   25,501번째 아이템에서 예외를 던집니다.
    //   청크 1,000 이면 → 25번째 청크까지 커밋 → settlement 25,000행.
    //   25,500 이 아니라 25,000 입니다. 롤백 단위가 아이템이 아니라 청크라서.
    // ========================================================================

    static class ExplodingProcessor implements ItemProcessor<Order, Settlement> {

        private final AtomicInteger seq = new AtomicInteger();
        private final ItemProcessor<Order, Settlement> delegate;
        private final int failAt;

        ExplodingProcessor(ItemProcessor<Order, Settlement> delegate, int failAt) {
            this.delegate = delegate;
            this.failAt = failAt;
        }

        @Override
        public Settlement process(Order item) throws Exception {
            int n = seq.incrementAndGet();
            if (n == failAt) {
                throw new IllegalStateException(
                        "의도적 실패: " + n + "번째 아이템 (order_id=" + item.order_id() + ")");
            }
            return delegate.process(item);
        }
        // ⚠️ AtomicInteger 카운터가 정확한 것은 이 Step 이 단일 스레드이기 때문입니다.
        //    Step 13 의 멀티스레드 Step 에 그대로 가져가면 25,501번째가 매번 달라집니다.
    }

    @Value("${batch.fail-at:25501}")
    private int failAt;

    @Bean
    public Step settlementFailStep(JobRepository jobRepository,
                                   PlatformTransactionManager txManager,
                                   ItemReader<Order> orderReader,
                                   ItemProcessor<Order, Settlement> settlementProcessor,
                                   ItemWriter<Settlement> settlementWriter) {
        return new StepBuilder("settlementFailStep", jobRepository)
                .<Order, Settlement>chunk(chunkSize, txManager)
                .reader(orderReader)
                .processor(new ExplodingProcessor(settlementProcessor, failAt))
                .writer(settlementWriter)
                .build();
    }

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

    // ========================================================================
    // [5-8] Chunk<T> API — 5.0 부터 write(List) 가 write(Chunk) 로 바뀌었습니다.
    //
    //   ⚠️ 이 writer 는 실제 INSERT 를 하지 않고 로그만 남깁니다.
    //      Chunk API 관찰용이므로, 이 writer 를 쓰는 Job 을 돌리면
    //      settlement 가 비어 있는 게 정상입니다.
    // ========================================================================

    static class SettlementLogWriter implements ItemWriter<Settlement> {

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

        @Override   // ← @Override 를 반드시 붙이세요. 4.x 의 write(List) 를
                    //   남겨 두면 컴파일러가 대신 잡아 줍니다 (본문 5-8 의 함정).
        public void write(Chunk<? extends Settlement> chunk) {

            // (1) Chunk 는 Iterable 이므로 향상된 for 문이 그대로 됩니다
            BigDecimal sum = BigDecimal.ZERO;
            for (Settlement s : chunk) {
                sum = sum.add(s.netAmount());
            }

            // (2) List 가 필요하면 getItems()
            List<? extends Settlement> items = chunk.getItems();

            // (3) size() / isEmpty()
            if (chunk.isEmpty()) {
                log.debug("빈 청크 — 마지막 사이클입니다");
                return;
            }

            log.debug("청크 {}건, 첫 주문 {}, 순액 합계 {}",
                    chunk.size(), items.get(0).orderId(), sum);
        }
    }

    @Bean
    public Step settlementLogStep(JobRepository jobRepository,
                                  PlatformTransactionManager txManager,
                                  ItemReader<Order> orderReader,
                                  ItemProcessor<Order, Settlement> settlementProcessor) {
        return new StepBuilder("settlementLogStep", jobRepository)
                .<Order, Settlement>chunk(chunkSize, txManager)
                .reader(orderReader)
                .processor(settlementProcessor)
                .writer(new SettlementLogWriter())
                .build();
    }

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

    // ========================================================================
    // [5-8] Chunk.of() — 테스트에서 Chunk 를 손쉽게 만드는 정적 팩터리
    //
    //   실제 테스트 클래스는 src/test 에 두지만, API 형태를 보여 주기 위해
    //   여기에 예시로 남깁니다.
    // ========================================================================

    static Chunk<Settlement> sampleChunk() {
        return Chunk.of(
                new Settlement(1L, 1, LocalDate.of(2025, 1, 1),
                        new BigDecimal("1000.00"), new BigDecimal("0.0300"),
                        new BigDecimal("30.00"), new BigDecimal("970.00")),
                new Settlement(2L, 2, LocalDate.of(2025, 1, 1),
                        new BigDecimal("2000.00"), new BigDecimal("0.0250"),
                        new BigDecimal("50.00"), new BigDecimal("1950.00"))
        );
    }

    // ========================================================================
    // [5-1] 참고 — 이렇게 하면 안 되는 Tasklet 방식.
    //   70,000건을 List 로 전부 들고 하나의 트랜잭션으로 처리합니다.
    //   메모리·트랜잭션·재시작 셋 다 문제입니다. 실행하지 마세요.
    // ========================================================================

    @SuppressWarnings("unused")
    static class DontDoThis {
        void badTasklet(JdbcTemplate jdbc) {
            List<Order> all = jdbc.query(
                    "SELECT order_id, customer_id, amount, status, ordered_at "
                            + "FROM orders WHERE status = 'COMPLETED'",
                    new DataClassRowMapper<>(Order.class));
            // 70,000개가 전부 힙에. 700만 건이면 OutOfMemoryError.
            // 그리고 69,999번째에서 실패하면 처음부터 다시.
            log.warn("{}건을 통째로 들고 있습니다", all.size());
        }
    }

    // 참고: BeanPropertyItemSqlParameterSourceProvider 는 record 에 안 되므로
    //       import 만 남겨 두고 쓰지 않습니다. Step 08 에서 이유를 다룹니다.
    @SuppressWarnings("unused")
    private static final Class<?> WHY_NOT_THIS = BeanPropertyItemSqlParameterSourceProvider.class;
}

package com.example.batch.step05;

/*
 * ============================================================================
 * Step 05 — 청크 지향 처리 / 연습문제 (6문제)
 * ============================================================================
 *
 * 정답은 Solution.java 에 있습니다. 먼저 직접 풀어 보세요.
 *
 * [매 실행 전 반드시 초기화]
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb <<'SQL'
 *   SET FOREIGN_KEY_CHECKS = 0;
 *   DELETE FROM BATCH_STEP_EXECUTION_CONTEXT;
 *   DELETE FROM BATCH_STEP_EXECUTION;
 *   DELETE FROM BATCH_JOB_EXECUTION_CONTEXT;
 *   DELETE FROM BATCH_JOB_EXECUTION_PARAMS;
 *   DELETE FROM BATCH_JOB_EXECUTION;
 *   DELETE FROM BATCH_JOB_INSTANCE;
 *   SET FOREIGN_KEY_CHECKS = 1;
 *   TRUNCATE TABLE settlement;
 *   SQL
 *
 *   초기화하지 않고 다시 돌리면 settlement.order_id 의 UNIQUE 제약에 걸려
 *   DuplicateKeyException 이 납니다. 이건 버그가 아니라 프로젝트 셋업이
 *   의도한 안전장치입니다 — 정산이 두 번 되느니 시끄럽게 실패하는 게 낫습니다.
 * ============================================================================
 */

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.List;

import com.example.batch.step05.Practice.Order;
import com.example.batch.step05.Practice.Settlement;

@Configuration
public class Exercise {

    // ========================================================================
    // 문제 1. 4.x 스타일 Step 정의를 5.x 로 마이그레이션하세요.
    //
    //   아래는 Spring Batch 4.3 예제 그대로입니다. 5.1.1 에서는 컴파일조차
    //   되지 않습니다(StepBuilderFactory 가 삭제됨).
    //
    //   [4.x 원본]
    //     @Bean
    //     public Step legacyStep() {
    //         return stepBuilderFactory.get("legacyStep")
    //                 .<Order, Settlement>chunk(1000)
    //                 .reader(orderReader)
    //                 .processor(settlementProcessor)
    //                 .writer(settlementWriter)
    //                 .build();
    //     }
    //
    //   힌트: 바꿔야 할 것이 두 가지입니다. 하나는 컴파일 에러로 드러나고,
    //         다른 하나는 컴파일이 통과한 뒤 런타임에 드러납니다.
    // ========================================================================

    @Bean
    public Step migratedStep(JobRepository jobRepository,
                             PlatformTransactionManager txManager,
                             ItemReader<Order> orderReader,
                             ItemProcessor<Order, Settlement> settlementProcessor,
                             ItemWriter<Settlement> settlementWriter) {

        // 여기에 작성:
        return null;
    }

    // ========================================================================
    // 문제 2. COMPLETED 주문 70,000건을 청크 크기 700 으로 처리하면
    //         BATCH_STEP_EXECUTION.COMMIT_COUNT 는 얼마일까요?
    //
    //   ⚠️ 먼저 종이에 답을 적고, 그다음 실행해서 검증하세요.
    //      실행 결과를 보고 역산하면 배우는 게 없습니다.
    //
    //   실행:
    //     ./gradlew bootRun --args='--spring.batch.job.name=settlementChunkJob --batch.chunk-size=700'
    //   검증:
    //     mysql -h127.0.0.1 -P3308 -ubatch -pbatch1234 batchdb -t -e "
    //       SELECT READ_COUNT, WRITE_COUNT, COMMIT_COUNT
    //       FROM BATCH_STEP_EXECUTION ORDER BY STEP_EXECUTION_ID DESC LIMIT 1;"
    //
    //   힌트: 70000 / 700 = 100 입니다. 그런데 답은 100 이 아닙니다.
    // ========================================================================

    /** 여기에 작성: 예측한 COMMIT_COUNT 값 */
    public static final int Q2_PREDICTED_COMMIT_COUNT = 0;

    /** 여기에 작성: 왜 그 값인지 한 줄 설명 */
    public static final String Q2_REASON = "";

    // ========================================================================
    // 문제 3. 청크 크기 300 으로 70,000건을 돌려 실행시간을 실측하세요.
    //         본문 5-6 의 U자 곡선에서 300 은 어디쯤일까요?
    //
    //   실행:
    //     ./gradlew bootRun --args='--spring.batch.job.name=settlementChunkJob --batch.chunk-size=300' \
    //       | grep 'executed in'
    //
    //   본문 실측값:
    //     청크   10 → 48.2초
    //     청크  100 → 12.7초
    //     청크 1000 →  6.1초   ← 최적
    //     청크 10000 →  7.4초
    //
    //   실측한 뒤, 300 이 "왼쪽 내리막(커밋 오버헤드 지배)" 인지
    //   "오른쪽 오르막(메모리·GC 지배)" 인지 판정하고 근거를 쓰세요.
    // ========================================================================

    /** 여기에 작성: 실측 실행시간 (초) */
    public static final double Q3_MEASURED_SECONDS = 0.0;

    /** 여기에 작성: "LEFT_SLOPE" 또는 "RIGHT_SLOPE" 와 그 근거 */
    public static final String Q3_VERDICT = "";

    // ========================================================================
    // 문제 4. 실패 지점을 정하고, settlement 에 남을 건수를 미리 계산하세요.
    //
    //   Practice.java 의 settlementFailJob 은 --batch.fail-at=N 으로
    //   N번째 아이템에서 예외를 던집니다.
    //
    //   조건: --batch.chunk-size=500, --batch.fail-at=  (여러분이 정하세요)
    //
    //   ① failAt 값을 하나 정합니다 (예: 12345)
    //   ② settlement 에 남을 행 수를 계산합니다  ← 실행 전에!
    //   ③ 실행해서 맞는지 확인합니다
    //
    //   실행:
    //     ./gradlew bootRun --args='--spring.batch.job.name=settlementFailJob \
    //       --batch.chunk-size=500 --batch.fail-at=12345'
    //   검증:
    //     mysql -h127.0.0.1 -P3308 -ubatch -pbatch1234 batchdb -t -e "
    //       SELECT COUNT(*) FROM settlement;"
    //
    //   힌트: 커밋된 청크 수를 먼저 나눗셈으로 구하고, 청크 크기를 곱합니다.
    //         나눗셈의 분자에 주의하세요. off-by-one 이 이 문제의 핵심입니다.
    // ========================================================================

    /** 여기에 작성: 여러분이 정한 failAt */
    public static final int Q4_FAIL_AT = 0;

    /** 여기에 작성: 청크 500 일 때 settlement 에 남을 행 수 */
    public static final int Q4_EXPECTED_ROWS = 0;

    /** 여기에 작성: 계산식 (예: "(N - 1) / 500 * 500") */
    public static final String Q4_FORMULA = "";

    // ========================================================================
    // 문제 5. 아래 4.x writer 를 5.x 로 고치세요.
    //
    //   ⚠️ 이 코드는 지금 상태로도 "컴파일이 됩니다."
    //      @Override 가 없어서, write(List) 는 그냥 이름이 같은 별개의
    //      메서드가 되고 인터페이스 쪽은 구현되지 않은 채로 남습니다.
    //      그 결과 배치는 COMPLETED 로 끝나는데 settlement 는 0행입니다.
    //
    //      "에러 없이 조용히 틀리는" 전형입니다. 고치세요.
    // ========================================================================

    static class LegacyWriter implements ItemWriter<Settlement> {

        private final JdbcTemplate jdbc;

        LegacyWriter(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        // ↓ 4.x 시그니처. 이대로 두면 조용히 아무것도 안 쓰입니다.
        public void write(List<? extends Settlement> items) {
            for (Settlement s : items) {
                jdbc.update("""
                            INSERT INTO settlement
                              (order_id, customer_id, settle_date, gross_amount,
                               fee_rate, fee_amount, net_amount)
                            VALUES (?, ?, ?, ?, ?, ?, ?)
                            """,
                        s.orderId(), s.customerId(), s.settleDate(),
                        s.grossAmount(), s.feeRate(), s.feeAmount(), s.netAmount());
            }
        }

        // 여기에 작성: 5.x 시그니처로 고친 write 메서드
        //             (@Override 를 반드시 붙이세요)
    }

    // ========================================================================
    // 문제 6. 안전한 최대 청크 크기를 계산하세요. (실행 불필요, 계산 문제)
    //
    //   조건:
    //     - JVM 힙 최대치      : 512 MB
    //     - 입력 아이템 하나   : 약 8 KB  (주문 + order_items 3건)
    //     - 출력 아이템 하나   : 약 4 KB  (정산 레코드)
    //     - Step 은 단일 스레드
    //
    //   본문 5-6 의 판단식:
    //     chunkSize × (입력 크기 + 출력 크기) × 2  ≤  힙 / 4
    //
    //   ① 위 식으로 상한을 구하세요.
    //   ② 실제로 고를 값을 정하고, 왜 상한보다 작게 잡는지 쓰세요.
    //   ③ 이 Step 을 4스레드로 바꾸면 상한이 어떻게 달라지나요?
    // ========================================================================

    /** 여기에 작성: ① 계산으로 구한 상한 */
    public static final int Q6_UPPER_BOUND = 0;

    /** 여기에 작성: ② 실제로 고를 청크 크기 */
    public static final int Q6_CHOSEN = 0;

    /** 여기에 작성: ② 왜 상한보다 작게 잡는가 */
    public static final String Q6_WHY_SMALLER = "";

    /** 여기에 작성: ③ 4스레드일 때의 상한 */
    public static final int Q6_UPPER_BOUND_4_THREADS = 0;

    // ========================================================================
    // 아래는 문제 1 의 Step 을 실행해 보기 위한 Job 입니다. 수정하지 마세요.
    // ========================================================================

    @Bean
    public Job exerciseJob(JobRepository jobRepository, Step migratedStep) {
        return new org.springframework.batch.core.job.builder.JobBuilder("exerciseJob", jobRepository)
                .start(migratedStep)
                .build();
    }

    /** 문제 5 를 고친 뒤, Chunk.of() 로 손쉽게 확인해 볼 수 있습니다. */
    @SuppressWarnings("unused")
    private static Chunk<Settlement> smokeTestChunk() {
        return Practice.sampleChunk();
    }
}

package com.example.batch.step05;

/*
 * ============================================================================
 * Step 05 — 청크 지향 처리 / 정답과 해설
 * ============================================================================
 * 문제를 직접 풀어 본 뒤에 여세요.
 * ============================================================================
 */

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
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.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.List;

import com.example.batch.step05.Practice.Order;
import com.example.batch.step05.Practice.Settlement;

@Configuration
public class Solution {

    // ========================================================================
    // 정답 1. 4.x → 5.x 마이그레이션
    // ========================================================================
    //
    // 바꿔야 할 것이 정확히 두 가지입니다.
    //
    //   (a) StepBuilderFactory 삭제  →  new StepBuilder(name, jobRepository)
    //       5.0 에서 JobBuilderFactory / StepBuilderFactory 가 모두 제거됐습니다.
    //       "Job 이름과 JobRepository 를 명시적으로 받는다"는 게 5.x 의 설계입니다.
    //       팩터리가 숨겨 주던 JobRepository 를 이제 개발자가 직접 넘깁니다.
    //       → 이건 컴파일 에러로 드러납니다. 안전한 변경입니다.
    //
    //   (b) .chunk(1000)  →  .chunk(1000, txManager)
    //       ← 이게 진짜 함정입니다.
    //
    //       5.x 에도 인자 하나짜리 chunk(int) 오버로드가 남아 있습니다.
    //       SimpleStepBuilder 를 직접 조립해 쓰는 경로를 위해 남겨 둔 것인데,
    //       그 바람에 4.x 코드를 복사해도 "컴파일이 통과합니다."
    //
    //       그리고 Step 빈이 만들어지는 시점에 이렇게 죽습니다.
    //
    //         java.lang.IllegalStateException: A transaction manager must be provided
    //           at org.springframework.util.Assert.state(Assert.java:76)
    //           at org.springframework.batch.core.step.builder.StepBuilderHelper
    //                 .enhance(StepBuilderHelper.java:198)
    //           at org.springframework.batch.core.step.builder.SimpleStepBuilder
    //                 .build(SimpleStepBuilder.java:145)
    //
    //       "컴파일이 됐으니 마이그레이션이 끝났다"고 판단하면 안 되는 이유입니다.
    //       5.x 로 옮길 때는 컴파일러가 아니라 **애플리케이션 기동**까지 확인하세요.
    //
    // 참고: 트랜잭션 매니저를 어디서 받는가?
    //   Boot 3.2 의 자동설정이 DataSourceTransactionManager 를 하나 만들어 둡니다.
    //   JPA 를 함께 쓰면 JpaTransactionManager 가 대신 들어옵니다.
    //   메타데이터 갱신과 업무 데이터 쓰기를 한 트랜잭션으로 묶으려면
    //   이 둘이 같은 DataSource 를 봐야 합니다 (프로젝트 셋업 P-1 참고).

    @Bean
    public Step migratedStep(JobRepository jobRepository,
                             PlatformTransactionManager txManager,
                             ItemReader<Order> orderReader,
                             ItemProcessor<Order, Settlement> settlementProcessor,
                             ItemWriter<Settlement> settlementWriter) {

        return new StepBuilder("migratedStep", jobRepository)   // (a)
                .<Order, Settlement>chunk(1000, txManager)      // (b)
                .reader(orderReader)
                .processor(settlementProcessor)
                .writer(settlementWriter)
                .build();
    }

    // ========================================================================
    // 정답 2. 청크 700 일 때의 COMMIT_COUNT = 101
    // ========================================================================
    //
    //   70,000 / 700 = 100 청크. 그런데 답은 101 입니다.
    //
    //   왜 하나 더인가:
    //     Spring Batch 는 ItemReader 가 null 을 반환해야만 "데이터가 끝났다"는
    //     것을 압니다. 이게 ItemReader 인터페이스의 계약입니다.
    //
    //       T read() throws Exception;   // 더 이상 없으면 null
    //
    //     100번째 청크가 70,000번째 아이템을 읽어 커밋해도, 프레임워크는
    //     아직 데이터가 남았는지 모릅니다. 리더에게 물어보지 않았으니까요.
    //     그래서 101번째 사이클을 한 번 더 돌려 read() 를 호출하고,
    //     null 을 받고, 빈 청크로 Step 을 종료합니다.
    //
    //     그 빈 트랜잭션도 커밋되므로 COMMIT_COUNT 가 1 늘어납니다.
    //     DEBUG 로그의 마지막 줄 `read=0` 이 바로 이 사이클입니다.
    //
    //   일반식:  COMMIT_COUNT = ceil(전체건수 / 청크크기) + 1
    //
    //     70,000 / 700   → 100 + 1 = 101
    //     70,000 / 1000  →  70 + 1 =  71
    //     70,000 / 300   → 234 + 1 = 235   (70000/300 = 233.33 → 올림 234)
    //     70,000 / 10000 →   7 + 1 =   8
    //
    //   실무에서 이게 왜 중요한가:
    //     "커밋 횟수로 처리량을 역산"하는 모니터링 대시보드를 만들 때
    //     이 +1 을 빼먹으면 항상 한 청크만큼 과대 집계됩니다.
    //     청크가 작을 때는 오차가 무시할 만하지만(7001 중 1),
    //     청크가 클 때는 8분의 1 — 12.5% 오차입니다.

    public static final int Q2_PREDICTED_COMMIT_COUNT = 101;
    public static final String Q2_REASON =
            "ceil(70000/700)=100 청크 + reader 가 null 을 반환하는 마지막 빈 사이클 1회";

    // ========================================================================
    // 정답 3. 청크 300 → 8.9초. LEFT_SLOPE (왼쪽 내리막)
    // ========================================================================
    //
    //   실측 (3회 중앙값):
    //     INFO --- o.s.batch.core.step.AbstractStep : Step: [settlementStep] executed in 8s914ms
    //
    //   본문 실측값과 나란히 놓으면:
    //
    //     청크    10 → 48.2초   (커밋 7,001회)
    //     청크   100 → 12.7초   (커밋   701회)
    //     청크   300 →  8.9초   (커밋   235회)   ← 여기
    //     청크  1000 →  6.1초   (커밋    71회)   ← 최적
    //     청크 10000 →  7.4초   (커밋     8회)
    //
    //   판정 근거 — LEFT_SLOPE 인 이유 두 가지:
    //
    //   (1) 청크를 300 → 1000 으로 더 키우면 여전히 빨라집니다 (8.9 → 6.1초).
    //       U자 곡선의 오른쪽에 있다면 키울수록 느려져야 합니다.
    //
    //   (2) GC 로그를 보면 메모리가 아직 병목이 아닙니다.
    //         [3.102s][info][gc] GC(4) Pause Young (Normal) 197M->72M(512M) 4.881ms
    //       pause 가 4.9ms 로, 청크 1000 의 7ms 보다도 작습니다.
    //       힙 peak 도 218MB 로 여유가 많습니다.
    //       즉 300 이 느린 건 메모리 때문이 아니라 **커밋 235회의 고정비용**
    //       때문입니다. 전형적인 왼쪽 내리막입니다.
    //
    //   덤으로 배울 점 — 수익 체감:
    //     10 → 100  : 48.2 → 12.7초  (35.5초 단축)
    //     100 → 300 : 12.7 →  8.9초  ( 3.8초 단축)
    //     300 → 1000:  8.9 →  6.1초  ( 2.8초 단축)
    //     1000 → 10000: 6.1 → 7.4초  (오히려 1.3초 손해)
    //
    //     청크 크기를 10배 키울 때마다 얻는 이득이 급격히 줄어듭니다.
    //     그래서 "일단 1,000 부터 시작하라"는 권고가 나옵니다 —
    //     1,000 근처에서는 어느 쪽으로 틀려도 손해가 작기 때문입니다.

    public static final double Q3_MEASURED_SECONDS = 8.9;
    public static final String Q3_VERDICT =
            "LEFT_SLOPE — 더 키우면 여전히 빨라지고(8.9→6.1초), "
            + "GC pause 4.9ms·힙 218MB 로 메모리는 아직 병목이 아님. 커밋 235회가 비용의 주범";

    // ========================================================================
    // 정답 4. failAt=12345, chunkSize=500 → settlement 12,000행
    // ========================================================================
    //
    //   계산식:   ((failAt - 1) / chunkSize) * chunkSize      ← 정수 나눗셈
    //
    //   대입:     ((12345 - 1) / 500) * 500
    //           = (12344 / 500) * 500
    //           = 24 * 500
    //           = 12,000
    //
    //   말로 풀면:
    //     12,345번째 아이템은 25번째 청크(12,001 ~ 12,500)에 속합니다.
    //     그 청크는 통째로 롤백되고, 1~24번 청크만 커밋됩니다.
    //     24 × 500 = 12,000행.
    //
    //   실측:
    //     +----------+
    //     | COUNT(*) |
    //     +----------+
    //     |    12000 |
    //     +----------+
    //
    //   ★ 여기서 핵심은 off-by-one 입니다. 경계를 직접 확인해 보세요.
    //
    //     failAt = 12500 → (12499 / 500) * 500 = 24 * 500 = 12,000
    //         12,500번째는 25번째 청크의 **마지막** 아이템입니다.
    //         그 청크가 롤백되므로 여전히 12,000행.
    //
    //     failAt = 12501 → (12500 / 500) * 500 = 25 * 500 = 12,500
    //         12,501번째는 26번째 청크의 **첫** 아이템입니다.
    //         25번째 청크는 이미 커밋됐으므로 12,500행.
    //
    //     분자가 failAt 이 아니라 (failAt - 1) 인 이유가 여기 있습니다.
    //     failAt / chunkSize 로 계산하면 12500/500 = 25 → 12,500 이라는
    //     틀린 답이 나옵니다. 딱 청크 하나(500건) 만큼 틀립니다.
    //
    //   실무로 옮기면:
    //     "배치가 실패했으니 12,345건까지는 처리됐겠지" 라고 가정하고
    //     12,346번째부터 수동 보정을 돌리면 **345건이 정산에서 누락**됩니다.
    //     실패한 배치의 복구 지점은 추측하지 말고
    //     BATCH_STEP_EXECUTION.WRITE_COUNT 를 읽으세요. 그 값이 12,000 입니다.

    public static final int Q4_FAIL_AT = 12345;
    public static final int Q4_EXPECTED_ROWS = 12000;
    public static final String Q4_FORMULA = "((failAt - 1) / chunkSize) * chunkSize  (정수 나눗셈)";

    // ========================================================================
    // 정답 5. write(List) → write(Chunk)
    // ========================================================================
    //
    //   고친 코드는 아래 FixedWriter 입니다. 변경 자체는 두 줄이지만,
    //   이 문제의 값어치는 코드가 아니라 **왜 이게 무서운가** 에 있습니다.
    //
    //   ── 무엇이 벌어지는가 ─────────────────────────────────────────────
    //
    //   ItemWriter<T> 는 5.x 에서 이렇게 생겼습니다.
    //
    //     public interface ItemWriter<T> {
    //         void write(Chunk<? extends T> chunk) throws Exception;
    //     }
    //
    //   추상 메서드가 하나뿐인 인터페이스인데, 문제의 LegacyWriter 는
    //   write(List<? extends Settlement>) 를 구현했습니다.
    //   시그니처가 다르므로 이건 **인터페이스 구현이 아니라 그냥 새 메서드**입니다.
    //
    //   그렇다면 왜 컴파일 에러가 안 났을까요?
    //   ItemWriter 의 추상 메서드가 구현되지 않았으니 에러가 나야 정상인데—
    //   실제로 인터페이스를 직접 implements 했다면 에러가 납니다.
    //   위험한 건 다음 세 경우입니다.
    //
    //     (1) @Override 없이 추상 클래스(예: AbstractItemStreamItemWriter)를 상속한 경우
    //         → 상위 클래스의 기본 구현이 대신 호출됩니다. 대개 아무 일도 안 합니다.
    //     (2) ItemWriter 를 람다/익명 클래스로 만들면서 위임 대상만 4.x 로 남긴 경우
    //     (3) 여러 writer 를 CompositeItemWriter 로 묶었는데 그중 하나만 4.x 인 경우
    //         → 나머지는 정상 동작하므로 "일부 데이터만 안 들어가는" 형태로 나타납니다.
    //
    //   ── 왜 이게 최악인가 ─────────────────────────────────────────────
    //
    //   결과가 이렇습니다.
    //
    //     STATUS       : COMPLETED     ← 배치는 성공했다고 말합니다
    //     READ_COUNT   : 70000
    //     WRITE_COUNT  : 70000         ← 카운터도 정상입니다
    //     settlement   : 0행           ← 그런데 데이터가 없습니다
    //
    //   WRITE_COUNT 는 "writer 에 넘긴 아이템 수"이지 "DB 에 들어간 행 수"가
    //   아니기 때문에, writer 가 아무것도 안 해도 70,000 으로 찍힙니다.
    //   모니터링도, 알림도, 후속 Step 의 검증도 전부 통과합니다.
    //
    //   ── 안전벨트 ─────────────────────────────────────────────────────
    //
    //   @Override 를 붙이면 컴파일러가 즉시 잡아 줍니다.
    //
    //     error: method does not override or implement a method from a supertype
    //         @Override
    //         ^
    //
    //   4.x → 5.x 마이그레이션에서 @Override 는 선택 사항이 아니라 필수입니다.
    //   "런타임에 조용히 틀리는 것"을 "컴파일 타임에 시끄럽게 실패하는 것"으로
    //   바꿔 주는 한 줄이기 때문입니다.
    //
    //   보너스: 실제 INSERT 여부를 검증하는 후속 Step 을 두는 것도 방법입니다.
    //           settlement 행 수와 WRITE_COUNT 를 대조하는 Tasklet 하나면 됩니다.

    static class FixedWriter implements ItemWriter<Settlement> {

        private final JdbcTemplate jdbc;

        FixedWriter(JdbcTemplate jdbc) {
            this.jdbc = jdbc;
        }

        @Override                                          // ← 안전벨트
        public void write(Chunk<? extends Settlement> chunk) {   // ← List 가 아니라 Chunk
            for (Settlement s : chunk) {                    // Chunk 는 Iterable
                jdbc.update("""
                            INSERT INTO settlement
                              (order_id, customer_id, settle_date, gross_amount,
                               fee_rate, fee_amount, net_amount)
                            VALUES (?, ?, ?, ?, ?, ?, ?)
                            """,
                        s.orderId(), s.customerId(), s.settleDate(),
                        s.grossAmount(), s.feeRate(), s.feeAmount(), s.netAmount());
            }
        }

        /**
         * 굳이 List 가 필요하다면 getItems() 로 꺼냅니다.
         * 다만 대부분의 경우 Chunk 를 그대로 순회하는 편이 낫습니다 —
         * 순회 중 remove() 가 가능한 ChunkIterator 를 쓸 수 있고,
         * getSkips() 같은 5.x 전용 정보에도 접근할 수 있기 때문입니다.
         */
        @SuppressWarnings("unused")
        private void alsoWorks(Chunk<? extends Settlement> chunk) {
            List<? extends Settlement> items = chunk.getItems();
            jdbc.batchUpdate("INSERT INTO settlement (order_id) VALUES (?)",
                    items.stream().map(s -> new Object[]{s.orderId()}).toList());
        }
    }

    // ========================================================================
    // 정답 6. 안전한 최대 청크 크기
    // ========================================================================
    //
    //   ① 상한 계산
    //
    //      판단식:  chunkSize × (입력 + 출력) × 2  ≤  힙 / 4
    //
    //      좌변의 ×2 는 "입력 N개와 출력 N개가 동시에 살아 있고,
    //      GC 가 회수하기 전 이전 청크의 잔재도 잠시 남는다"는 여유분입니다.
    //      우변의 /4 는 "힙 전체를 청크 하나에 쓸 수는 없다"는 뜻입니다.
    //      나머지 3/4 은 커넥션 풀, Hibernate 1차 캐시, 프레임워크, 그리고
    //      GC 가 일할 여유 공간(headroom)이 씁니다.
    //
    //        입력 + 출력 = 8 KB + 4 KB = 12 KB
    //        힙 / 4      = 512 MB / 4 = 128 MB = 131,072 KB
    //
    //        chunkSize ≤ 131,072 KB / (12 KB × 2)
    //                  = 131,072 / 24
    //                  = 5,461.3
    //                  → 5,461
    //
    //   ② 실제로 고를 값: 1,000
    //
    //      상한 5,461 은 "여기까지는 OOM 이 안 난다"는 선일 뿐이고,
    //      "여기가 가장 빠르다"는 뜻이 전혀 아닙니다.
    //      본문 5-6 의 실측이 정확히 이 점을 보여 줍니다 —
    //      청크 10,000 은 OOM 없이 완주했지만 청크 1,000 보다 느렸습니다.
    //      GC pause 가 7ms 에서 44ms 로 늘었기 때문입니다.
    //
    //      상한 근처를 고른다는 건 **GC 를 감수하겠다**는 선택입니다.
    //      게다가 다음 위험이 남습니다.
    //
    //        - 아이템 크기 8KB 는 평균입니다. 실제 데이터에는 꼬리가 있습니다.
    //          주문 하나에 order_items 가 3건이 아니라 300건인 이상치가
    //          섞여 있으면 그 청크만 폭발합니다.
    //        - 데이터는 늘어납니다. 오늘 8KB 인 아이템이 컬럼 추가로
    //          내년에 12KB 가 되면 상한이 3,640 으로 내려앉습니다.
    //        - 힙 512MB 는 컨테이너 메모리 한도보다 작아야 합니다.
    //          JVM 은 힙 외에 메타스페이스·스레드 스택·다이렉트 버퍼도 씁니다.
    //
    //      1,000 은 상한의 약 1/5 로, 위 세 가지가 동시에 나빠져도
    //      버틸 여유가 있습니다. 그리고 실측상 성능도 최적 구간입니다.
    //
    //   ③ 4스레드로 바꾸면
    //
    //      멀티스레드 Step(Step 13)에서는 각 스레드가 자기 청크를 들고 있습니다.
    //      즉 동시에 살아 있는 아이템이 4배가 됩니다.
    //
    //        chunkSize ≤ 5,461 / 4 = 1,365
    //
    //      실제로 고를 값은 여기서도 여유를 두어 250 ~ 500 이 적당합니다.
    //      "스레드를 4배로 늘렸으니 청크도 4배로" 는 정확히 반대입니다.
    //      스레드를 늘리면 청크는 **줄여야** 합니다.
    //
    //      덧붙여, 멀티스레드 Step 은 청크 크기 × 스레드 수만큼의
    //      DB 커넥션도 동시에 씁니다. application.yml 의
    //      hikari.maximum-pool-size: 20 이 그래서 필요합니다.

    public static final int Q6_UPPER_BOUND = 5461;
    public static final int Q6_CHOSEN = 1000;
    public static final String Q6_WHY_SMALLER =
            "상한은 OOM 이 안 나는 선일 뿐 최적점이 아니다. 상한 근처는 GC pause 를 감수하는 선택이고, "
            + "아이템 크기의 꼬리(이상치)·데이터 증가·힙 외 메모리까지 감안하면 상한의 1/5 이 안전하다. "
            + "실측상 1,000 이 성능 최적 구간이기도 하다.";
    public static final int Q6_UPPER_BOUND_4_THREADS = 1365;

    // ========================================================================
    // 실행용 Job
    // ========================================================================

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