#Step 13 — 병렬 처리와 확장

@Bean
public Step settlementSingleStep(JobRepository jobRepository,
                                 PlatformTransactionManager txManager,
                                 DataSource dataSource) {
    return new StepBuilder("settlementSingleStep", jobRepository)
            .<Order, Settlement>chunk(1000, txManager)
            .reader(orderCursorReader(dataSource))
            .processor(new SettlementProcessor())
            .writer(settlementWriter(dataSource))
            .build();
}

@Bean
public JdbcCursorItemReader<Order> orderCursorReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Order>()
            .name("orderCursorReader")
            .dataSource(dataSource)
            .fetchSize(1000)
            .sql("""
                 SELECT order_id, customer_id, amount, status, ordered_at
                 FROM orders
                 WHERE status = 'COMPLETED'
                 ORDER BY order_id
                 """)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

@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)
                 ON DUPLICATE KEY UPDATE
                   fee_amount = VALUES(fee_amount),
                   net_amount = VALUES(net_amount)
                 """)
            .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();
}

@Bean
public TaskExecutor batchTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setMaxPoolSize(4);
    executor.setQueueCapacity(0);                 // 큐에 쌓지 않고 바로 스레드로
    executor.setThreadNamePrefix("batch-mt-");
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.initialize();
    return executor;
}

@Bean
public Step settlementMultiThreadStep(JobRepository jobRepository,
                                      PlatformTransactionManager txManager,
                                      DataSource dataSource,
                                      TaskExecutor batchTaskExecutor) {
    return new StepBuilder("settlementMultiThreadStep", jobRepository)
            .<Order, Settlement>chunk(1000, txManager)
            .reader(orderCursorReader(dataSource))      // 13-0 의 그 Reader 를 그대로
            .processor(new SettlementProcessor())
            .writer(settlementWriter(dataSource))
            .taskExecutor(batchTaskExecutor)            // ← 이 한 줄
            .build();
}

// JdbcCursorItemReader 내부 (개념 코드)
protected Order doRead() throws Exception {
    if (rs.next()) {                   // ① 커서를 한 칸 전진
        return rowMapper.mapRow(rs, currentRow++);   // ② 지금 위치의 행을 매핑
    }
    return null;
}

// ⚠️ 절대 멀티스레드 Step 에 쓰지 마세요
public class NaiveOrderReader implements ItemReader<Order> {
    private final List<Order> orders;
    private int index = 0;                       // ← 이 필드 하나가 전부입니다

    @Override
    public Order read() {
        if (index < orders.size()) {
            return orders.get(index++);          // index++ 는 원자적이지 않습니다
        }
        return null;
    }
}

@Bean
public SynchronizedItemStreamReader<Order> syncOrderReader(DataSource dataSource) {
    SynchronizedItemStreamReader<Order> reader = new SynchronizedItemStreamReader<>();
    reader.setDelegate(orderCursorReader(dataSource));
    return reader;
}

return new SynchronizedItemStreamReaderBuilder<Order>()
        .delegate(orderCursorReader(dataSource))
        .build();

// SynchronizedItemStreamReader 내부
private final Lock lock = new ReentrantLock();

@Override
public T read() throws Exception {
    this.lock.lock();
    try {
        return this.delegate.read();     // ①과 ②를 하나의 임계 구역으로 묶는다
    } finally {
        this.lock.unlock();
    }
}

executor.setCorePoolSize(8);
executor.setMaxPoolSize(8);

@Bean
public JdbcCursorItemReader<Order> orderCursorReaderNoState(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Order>()
            .name("orderCursorReaderNoState")
            .dataSource(dataSource)
            .fetchSize(1000)
            .saveState(false)            // ← 재시작 정보를 저장하지 않습니다
            .sql("""
                 SELECT order_id, customer_id, amount, status, ordered_at
                 FROM orders WHERE status = 'COMPLETED' ORDER BY order_id
                 """)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

@Bean
public Job parallelStepJob(JobRepository jobRepository,
                           Step settlementSingleStep,
                           Step customerStatStep,
                           Step mergeReportStep) {

    Flow settlementFlow = new FlowBuilder<SimpleFlow>("settlementFlow")
            .start(settlementSingleStep)
            .build();

    Flow statFlow = new FlowBuilder<SimpleFlow>("statFlow")
            .start(customerStatStep)
            .build();

    Flow splitFlow = new FlowBuilder<SimpleFlow>("splitFlow")
            .start(settlementFlow)
            .split(new SimpleAsyncTaskExecutor("split-"))   // ← 여기서 갈라집니다
            .add(statFlow)
            .build();

    return new JobBuilder("parallelStepJob", jobRepository)
            .start(splitFlow)
            .next(mergeReportStep)          // 둘 다 끝나야 실행됩니다
            .end()
            .build();
}

public class OrderIdRangePartitioner implements Partitioner {

    private final JdbcTemplate jdbcTemplate;

    public OrderIdRangePartitioner(DataSource dataSource) {
        this.jdbcTemplate = new JdbcTemplate(dataSource);
    }

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        Long min = jdbcTemplate.queryForObject(
                "SELECT MIN(order_id) FROM orders WHERE status = 'COMPLETED'", Long.class);
        Long max = jdbcTemplate.queryForObject(
                "SELECT MAX(order_id) FROM orders WHERE status = 'COMPLETED'", Long.class);

        long targetSize = (max - min) / gridSize + 1;

        Map<String, ExecutionContext> result = new LinkedHashMap<>();
        long start = min;
        for (int i = 0; i < gridSize; i++) {
            ExecutionContext ctx = new ExecutionContext();
            long end = Math.min(start + targetSize - 1, max);
            ctx.putLong("minId", start);
            ctx.putLong("maxId", end);
            ctx.putString("partitionName", "partition" + i);
            result.put("partition" + i, ctx);
            start = end + 1;
        }
        return result;
    }
}

@Bean
public Step settlementPartitionMasterStep(JobRepository jobRepository,
                                          Step settlementWorkerStep,
                                          DataSource dataSource,
                                          TaskExecutor partitionTaskExecutor) {
    return new StepBuilder("settlementPartitionMasterStep", jobRepository)
            .partitioner("settlementWorkerStep", new OrderIdRangePartitioner(dataSource))
            .step(settlementWorkerStep)
            .partitionHandler(partitionHandler(settlementWorkerStep, partitionTaskExecutor))
            .build();
}

@Bean
public TaskExecutorPartitionHandler partitionHandler(Step settlementWorkerStep,
                                                     TaskExecutor partitionTaskExecutor) {
    TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
    handler.setStep(settlementWorkerStep);
    handler.setTaskExecutor(partitionTaskExecutor);
    handler.setGridSize(8);                 // Partitioner.partition(8) 로 전달됩니다
    return handler;
}

@Bean
@StepScope                                          // ← 없으면 동작하지 않습니다
public JdbcPagingItemReader<Order> partitionedOrderReader(
        DataSource dataSource,
        @Value("#{stepExecutionContext['minId']}") Long minId,
        @Value("#{stepExecutionContext['maxId']}") Long maxId) {

    MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
    provider.setSelectClause("order_id, customer_id, amount, status, ordered_at");
    provider.setFromClause("FROM orders");
    provider.setWhereClause("WHERE status = 'COMPLETED' AND order_id BETWEEN :minId AND :maxId");
    provider.setSortKeys(Map.of("order_id", Order.ASCENDING));

    return new JdbcPagingItemReaderBuilder<Order>()
            .name("partitionedOrderReader")
            .dataSource(dataSource)
            .queryProvider(provider)
            .parameterValues(Map.of("minId", minId, "maxId", maxId))
            .pageSize(1000)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

@Bean
public Step settlementWorkerStep(JobRepository jobRepository,
                                 PlatformTransactionManager txManager,
                                 JdbcPagingItemReader<Order> partitionedOrderReader,
                                 JdbcBatchItemWriter<Settlement> settlementWriter) {
    return new StepBuilder("settlementWorkerStep", jobRepository)
            .<Order, Settlement>chunk(1000, txManager)
            .reader(partitionedOrderReader)
            .processor(new SettlementProcessor())
            .writer(settlementWriter)
            .build();
}

// 마스터: TaskExecutorPartitionHandler 대신 MessageChannelPartitionHandler
@Bean
public MessageChannelPartitionHandler remotePartitionHandler(
        MessagingTemplate messagingTemplate, JobExplorer jobExplorer) {
    MessageChannelPartitionHandler handler = new MessageChannelPartitionHandler();
    handler.setStepName("settlementWorkerStep");
    handler.setGridSize(8);
    handler.setMessagingOperations(messagingTemplate);   // 요청 큐로 전송
    handler.setJobExplorer(jobExplorer);                 // 완료는 DB 폴링으로 확인
    handler.setPollInterval(2000L);
    return handler;
}

package com.example.batch.step13;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.item.support.SynchronizedItemStreamReader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Step 13 — 병렬 처리와 확장 : 본문 13-0 ~ 13-11 의 모든 예제.
 *
 * ─────────────────────────────────────────────────────────────────────────
 * 실행 방법
 *
 *   ./gradlew bootRun -Pargs=--spring.batch.job.name=settlementSingleJob
 *
 * 바깥 클래스에는 @Configuration 이 없습니다. 아래 static class 중 지금 측정할
 * 것 하나만 @Configuration 주석을 풀고 나머지는 주석 처리한 채로 돌리십시오.
 * 전부 켜 두면 Job 빈이 우르르 등록되어 측정이 뒤섞입니다.
 *
 * ─────────────────────────────────────────────────────────────────────────
 * ⚠️ 매 실행 전에 반드시 RESET_SQL 을 돌리십시오.
 *
 * 이 스텝에서 가장 흔한 실습 실수가 이것을 빼먹는 것입니다. settlement 에
 * 이전 실행의 70,000행이 그대로 남아 있으면, 이번 실행이 69,213건만 썼어도
 * 최종 COUNT 는 70,000 으로 보입니다. UNIQUE(order_id) 가 중복을 흡수하기
 * 때문입니다. 즉 "누락을 발견하지 못하게" 됩니다 — 13-3 의 핵심이 통째로
 * 무의미해집니다.
 * ─────────────────────────────────────────────────────────────────────────
 */
public class Practice {

    /** 매 측정 전에 실행할 초기화 SQL. */
    public static final String RESET_SQL = """
            -- mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "
            TRUNCATE TABLE settlement;
            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;
            -- "
            """;

    /** 측정 후 검증 SQL. distinct_orders 가 70000 이 아니면 무언가 잘못된 것입니다. */
    public static final String VERIFY_SQL = """
            SELECT COUNT(*)                  AS rows_written,
                   COUNT(DISTINCT order_id)  AS distinct_orders,
                   SUM(net_amount)           AS total_net
            FROM settlement;
            """;

    // =====================================================================
    // 공통 — 모든 구성이 같은 Processor / Writer 를 씁니다.
    // 비교 측정이 목적이므로 Reader 이외의 변수는 고정해야 합니다.
    // =====================================================================

    /**
     * 등급별 수수료를 적용해 Order → Settlement 로 변환합니다.
     *
     * 상태가 없습니다(필드 없음). 이것이 이 스텝 전체의 전제입니다.
     * 연습문제 2 의 RunningTotalProcessor 는 일부러 이 원칙을 깨뜨립니다.
     */
    public static class SettlementProcessor implements ItemProcessor<Order, Settlement> {

        /** customer_id % 4 → 등급 수수료율. 시드 규칙과 일치시킵니다. */
        private static final BigDecimal[] FEE_RATES = {
                new BigDecimal("0.0350"),   // n % 4 == 0 → BRONZE
                new BigDecimal("0.0300"),   // 1 → SILVER
                new BigDecimal("0.0250"),   // 2 → GOLD
                new BigDecimal("0.0200")    // 3 → VIP
        };

        @Override
        public Settlement process(Order order) {
            BigDecimal feeRate = FEE_RATES[order.customerId() % 4];
            BigDecimal gross = order.amount();
            BigDecimal fee = gross.multiply(feeRate).setScale(2, RoundingMode.HALF_UP);
            BigDecimal net = gross.subtract(fee);

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

    /**
     * 멱등 Writer — INSERT ... ON DUPLICATE KEY UPDATE.
     *
     * settlement.order_id 의 UNIQUE 제약 덕분에 같은 주문을 두 번 써도 행이
     * 늘어나지 않습니다. 재시작에 안전합니다.
     *
     * ⚠️ 그런데 13-3 에서 이 멱등성이 오히려 독이 됩니다.
     *    중복은 흡수하지만 "누락"은 전혀 감지하지 못하기 때문입니다.
     */
    public static JdbcBatchItemWriter<Settlement> buildSettlementWriter(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)
                     ON DUPLICATE KEY UPDATE
                       gross_amount = VALUES(gross_amount),
                       fee_amount   = VALUES(fee_amount),
                       net_amount   = VALUES(net_amount)
                     """)
                // record 는 자바빈이 아니므로 beanMapped() 를 쓸 수 없습니다.
                // Step 08 에서 다룬 그 이유입니다. 람다로 직접 매핑합니다.
                .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();
    }

    // =====================================================================
    // [13-0] 기준선 — 단일 스레드. 62.4초.
    //
    // 다른 어떤 절보다 먼저 실행하십시오. 비교 대상 없이 "18.9초가 나왔다"는
    // 아무 의미가 없습니다. 여러분의 머신에서의 기준선을 먼저 확보하십시오.
    // =====================================================================

    // @Configuration
    public static class Baseline {

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

        @Bean
        public Step settlementSingleStep(JobRepository jobRepository,
                                         PlatformTransactionManager txManager,
                                         DataSource dataSource) {
            return new StepBuilder("settlementSingleStep", jobRepository)
                    .<Order, Settlement>chunk(1000, txManager)
                    .reader(orderCursorReader(dataSource))
                    .processor(new SettlementProcessor())
                    .writer(buildSettlementWriter(dataSource))
                    .build();
        }

        /**
         * JdbcCursorItemReader — 커넥션을 하나 잡고 결과를 스트리밍합니다.
         *
         * 단일 스레드에서는 가장 빠른 선택입니다. 페이징처럼 매번 새 쿼리를
         * 날리지 않기 때문입니다.
         *
         * ⚠️ 그러나 이 Reader 는 스레드 안전하지 않습니다. 13-2 에서 바로
         *    이 Reader 를 멀티스레드 Step 에 그대로 넣습니다.
         */
        @Bean
        public JdbcCursorItemReader<Order> orderCursorReader(DataSource dataSource) {
            return new JdbcCursorItemReaderBuilder<Order>()
                    .name("orderCursorReader")
                    .dataSource(dataSource)
                    .fetchSize(1000)
                    .sql("""
                         SELECT order_id, customer_id, amount, status, ordered_at
                         FROM orders
                         WHERE status = 'COMPLETED'
                         ORDER BY order_id
                         """)
                    .rowMapper(new DataClassRowMapper<>(Order.class))
                    .build();
        }
    }

    // =====================================================================
    // [13-2] 멀티스레드 Step — .taskExecutor(...) 한 줄
    //
    // 18.9초. 3.3배 빨라집니다. 그리고 787건을 잃습니다.
    // =====================================================================

    // @Configuration
    public static class MultiThread {

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

        @Bean
        public Step settlementMultiThreadStep(JobRepository jobRepository,
                                              PlatformTransactionManager txManager,
                                              DataSource dataSource,
                                              TaskExecutor batchTaskExecutor) {
            return new StepBuilder("settlementMultiThreadStep", jobRepository)
                    .<Order, Settlement>chunk(1000, txManager)
                    .reader(new Baseline().orderCursorReader(dataSource))   // ← 공유됩니다!
                    .processor(new SettlementProcessor())
                    .writer(buildSettlementWriter(dataSource))
                    .taskExecutor(batchTaskExecutor)        // ← 이 한 줄이 전부입니다
                    .build();
        }

        /**
         * 스레드 4개.
         *
         * 커넥션 풀이 20 이므로 4개는 여유롭습니다. 13-9 에서 이 숫자를
         * 올렸을 때 무슨 일이 벌어지는지 측정합니다.
         */
        @Bean
        public TaskExecutor batchTaskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(4);
            executor.setMaxPoolSize(4);
            executor.setThreadNamePrefix("batch-worker-");
            executor.initialize();
            return executor;
        }
    }

    // =====================================================================
    // [13-3] 함정 재현 — 직접 만든 Reader 로 원인을 눈에 보이게
    //
    // JdbcCursorItemReader 의 내부는 복잡해서 "왜 사라졌는지"가 잘 안 보입니다.
    // 같은 결함을 가진 최소 코드를 직접 만들어 원인을 드러냅니다.
    //
    // ⚠️ 이 클래스는 "일부러 틀린 코드"입니다. index++ 를 고치지 마십시오.
    //    틀린 것을 눈으로 보는 것이 이 절의 목적입니다.
    // =====================================================================

    // @Configuration
    public static class ThreadUnsafe {

        /**
         * 스레드 안전하지 않은 Reader.
         *
         * 결함은 딱 한 줄, `return orders.get(index++)` 입니다.
         *
         * index++ 는 원자적이지 않습니다. 세 단계로 나뉩니다.
         *   ① index 를 읽는다
         *   ② 1을 더한다
         *   ③ 다시 쓴다
         *
         * 스레드 A 와 B 가 동시에 ①을 실행하면 둘 다 같은 값(예: 500)을 읽고,
         * 둘 다 500번 아이템을 처리한 뒤 index 를 501로 만듭니다.
         *   → 500번 아이템은 두 번 처리되고 (UNIQUE 가 흡수해서 안 보임)
         *   → 501번이 되었어야 할 자리가 밀려 하나가 통째로 누락됩니다.
         *
         * 이것이 787건이 사라진 이유입니다. 그리고 스레드 스케줄링에 의존하므로
         * 매 실행마다 손실 건수가 다릅니다 (69213 / 68940 / 69551).
         */
        public static class NaiveOrderReader implements ItemReader<Order> {

            private final List<Order> orders;
            private int index = 0;          // ← 여러 스레드가 공유하는 가변 상태

            public NaiveOrderReader(List<Order> orders) {
                this.orders = orders;
            }

            @Override
            public Order read() {
                if (index >= orders.size()) {
                    return null;            // null = 더 읽을 것 없음
                }
                return orders.get(index++); // ← 원자적이지 않습니다
            }
        }

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

        @Bean
        public Step settlementNaiveStep(JobRepository jobRepository,
                                        PlatformTransactionManager txManager,
                                        DataSource dataSource,
                                        TaskExecutor batchTaskExecutor) {
            List<Order> all = loadAllCompletedOrders(dataSource);
            return new StepBuilder("settlementNaiveStep", jobRepository)
                    .<Order, Settlement>chunk(1000, txManager)
                    .reader(new NaiveOrderReader(all))
                    .processor(new SettlementProcessor())
                    .writer(buildSettlementWriter(dataSource))
                    .taskExecutor(batchTaskExecutor)
                    .build();
        }

        /** 70,000건을 메모리에 통째로 올립니다. 실습 전용입니다. */
        static List<Order> loadAllCompletedOrders(DataSource dataSource) {
            return new JdbcTemplate(dataSource).query("""
                    SELECT order_id, customer_id, amount, status, ordered_at
                    FROM orders WHERE status = 'COMPLETED' ORDER BY order_id
                    """, new DataClassRowMapper<>(Order.class));
        }
    }

    // =====================================================================
    // [13-4] SynchronizedItemStreamReader — 해법과 그 천장
    //
    // 21.6초. 정확해지지만 단일 스레드 대비 2.9배에서 멈춥니다.
    // 읽기가 직렬화되기 때문입니다.
    // =====================================================================

    // @Configuration
    public static class Synchronized {

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

        @Bean
        public Step settlementSyncStep(JobRepository jobRepository,
                                       PlatformTransactionManager txManager,
                                       DataSource dataSource,
                                       TaskExecutor batchTaskExecutor) {
            return new StepBuilder("settlementSyncStep", jobRepository)
                    .<Order, Settlement>chunk(1000, txManager)
                    .reader(synchronizedOrderReader(dataSource))
                    .processor(new SettlementProcessor())
                    .writer(buildSettlementWriter(dataSource))
                    .taskExecutor(batchTaskExecutor)
                    .build();
        }

        /**
         * read() 전체를 락으로 감쌉니다.
         *
         * 장점: 어떤 ItemStreamReader 에도 적용됩니다. 범용 해법입니다.
         * 한계: 읽기가 완전히 직렬화됩니다. 스레드를 8개, 16개로 늘려도
         *       읽는 속도는 그대로라 3~4배가 천장입니다.
         *
         * 이 천장을 넘으려면 "읽기 자체를 나눠야" 합니다 → 파티셔닝(13-7).
         */
        @Bean
        public SynchronizedItemStreamReader<Order> synchronizedOrderReader(DataSource dataSource) {
            SynchronizedItemStreamReader<Order> reader = new SynchronizedItemStreamReader<>();
            reader.setDelegate(new Baseline().orderCursorReader(dataSource));
            return reader;
        }
    }

    // =====================================================================
    // [13-5] saveState(false) — 속도를 위해 재시작을 포기한다
    //
    // 멀티스레드 Step 에서는 사실상 필수입니다. 여러 스레드가 하나의
    // ExecutionContext 에 "어디까지 읽었는지"를 동시에 쓰면 그 값 자체가
    // 신뢰할 수 없게 되기 때문입니다.
    //
    // 저장된 값이 틀리면 재시작이 "틀린 지점"에서 재개됩니다. 차라리
    // 저장하지 않는 편이 정직합니다.
    // =====================================================================

    // @Configuration
    public static class NoSaveState {

        @Bean
        public JdbcCursorItemReader<Order> noStateOrderReader(DataSource dataSource) {
            return new JdbcCursorItemReaderBuilder<Order>()
                    .name("noStateOrderReader")
                    .dataSource(dataSource)
                    .fetchSize(1000)
                    .saveState(false)       // ← 재시작 포기 선언
                    .sql("""
                         SELECT order_id, customer_id, amount, status, ordered_at
                         FROM orders WHERE status = 'COMPLETED' ORDER BY order_id
                         """)
                    .rowMapper(new DataClassRowMapper<>(Order.class))
                    .build();
        }
    }

    // =====================================================================
    // [13-6] 병렬 Step (split) — 서로 다른 일을 동시에
    //
    // 63.0초. 1.1배밖에 안 빨라집니다.
    //
    // split 은 "같은 일을 나눠서" 하는 게 아니라 "다른 일을 동시에" 하는
    // 것입니다. 그래서 가장 긴 Step 보다 빨라질 수 없습니다.
    // 정산(62초) + 통계(6초) 를 병렬로 돌려도 62초입니다.
    // =====================================================================

    // @Configuration
    public static class ParallelSteps {

        @Bean
        public Job settlementSplitJob(JobRepository jobRepository,
                                      Step settlementSingleStep,
                                      Step dailyStatsStep) {
            return new JobBuilder("settlementSplitJob", jobRepository)
                    .start(new org.springframework.batch.core.job.builder.FlowBuilder<
                                    org.springframework.batch.core.job.flow.Flow>("splitFlow")
                            .split(new SimpleAsyncTaskExecutor("split-"))
                            .add(
                                    new org.springframework.batch.core.job.builder.FlowBuilder<
                                            org.springframework.batch.core.job.flow.Flow>("f1")
                                            .start(settlementSingleStep).build(),
                                    new org.springframework.batch.core.job.builder.FlowBuilder<
                                            org.springframework.batch.core.job.flow.Flow>("f2")
                                            .start(dailyStatsStep).build())
                            .build())
                    .end()
                    .build();
        }

        /** 정산과 무관한 별도 집계. 약 6초 걸립니다. */
        @Bean
        public Step dailyStatsStep(JobRepository jobRepository,
                                   PlatformTransactionManager txManager,
                                   DataSource dataSource) {
            return new StepBuilder("dailyStatsStep", jobRepository)
                    .tasklet((contribution, chunkContext) -> {
                        new JdbcTemplate(dataSource).query("""
                                SELECT DATE(ordered_at) d, COUNT(*) c, SUM(amount) s
                                FROM orders WHERE status = 'COMPLETED'
                                GROUP BY DATE(ordered_at) ORDER BY d
                                """, rs -> {
                            // 실습에서는 집계 결과를 소비만 합니다.
                        });
                        return org.springframework.batch.repeat.RepeatStatus.FINISHED;
                    }, txManager)
                    .build();
        }
    }

    // =====================================================================
    // [13-7] [13-8] 파티셔닝 — 데이터를 나누고 Step 을 복제한다
    //
    // 11.2초. 5.6배. 그리고 재시작까지 됩니다.
    //
    // 핵심은 "공유 상태가 없다"는 것입니다. 파티션마다 자기만의 Reader 를
    // 새로 만들어 쓰기 때문에 13-3 의 index++ 문제가 원천적으로 없습니다.
    // =====================================================================

    /**
     * order_id 범위를 gridSize 조각으로 나눕니다.
     *
     * 우리 데이터는 MIN=1, MAX=100000 이고 COMPLETED 가 order_id % 10 <= 6
     * 규칙으로 균등하게 깔려 있어, gridSize=8 이면 파티션당 정확히 8,750건이
     * 됩니다. 70000 / 8 = 8750.
     *
     * ⚠️ 이 균등함은 실습 데이터가 결정론적이라 얻어진 행운입니다.
     *    실제 데이터에서는 반드시 스큐가 생깁니다 — 연습문제 3 참고.
     */
    public static class OrderIdRangePartitioner implements Partitioner {

        private final JdbcTemplate jdbcTemplate;

        public OrderIdRangePartitioner(DataSource dataSource) {
            this.jdbcTemplate = new JdbcTemplate(dataSource);
        }

        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            Long min = jdbcTemplate.queryForObject(
                    "SELECT MIN(order_id) FROM orders WHERE status = 'COMPLETED'", Long.class);
            Long max = jdbcTemplate.queryForObject(
                    "SELECT MAX(order_id) FROM orders WHERE status = 'COMPLETED'", Long.class);

            long targetSize = (max - min) / gridSize + 1;

            Map<String, ExecutionContext> result = new LinkedHashMap<>();
            long start = min;
            for (int i = 0; i < gridSize; i++) {
                ExecutionContext ctx = new ExecutionContext();
                long end = Math.min(start + targetSize - 1, max);
                ctx.putLong("minId", start);
                ctx.putLong("maxId", end);
                ctx.putString("partitionName", "partition" + i);
                result.put("partition" + i, ctx);
                start = end + 1;
            }
            return result;
        }
    }

    // @Configuration
    public static class Partitioning {

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

        @Bean
        public Step settlementPartitionMasterStep(JobRepository jobRepository,
                                                  Step settlementWorkerStep,
                                                  DataSource dataSource,
                                                  TaskExecutor partitionTaskExecutor) {
            return new StepBuilder("settlementPartitionMasterStep", jobRepository)
                    .partitioner("settlementWorkerStep", new OrderIdRangePartitioner(dataSource))
                    .step(settlementWorkerStep)
                    .partitionHandler(partitionHandler(settlementWorkerStep, partitionTaskExecutor))
                    .build();
        }

        @Bean
        public TaskExecutorPartitionHandler partitionHandler(Step settlementWorkerStep,
                                                             TaskExecutor partitionTaskExecutor) {
            TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
            handler.setStep(settlementWorkerStep);
            handler.setTaskExecutor(partitionTaskExecutor);
            // ⚠️ gridSize 는 "스레드 수"가 아니라 "조각 수"입니다.
            //    동시에 몇 개가 도는지는 partitionTaskExecutor 의 풀 크기가 정합니다.
            handler.setGridSize(8);
            return handler;
        }

        @Bean
        public TaskExecutor partitionTaskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(8);
            executor.setMaxPoolSize(8);
            executor.setThreadNamePrefix("partition-");
            executor.initialize();
            return executor;
        }

        /**
         * [13-8] 늦은 바인딩.
         *
         * @StepScope 가 없으면 이 메서드는 기동 시점에 딱 한 번 호출되고,
         * 그때는 stepExecutionContext 가 존재하지 않습니다.
         *
         * 연습문제 4: 아래 @StepScope 한 줄만 주석 처리하고 돌려 보십시오.
         * 어떤 예외가 "어느 시점에" 나는지가 문제의 핵심입니다.
         */
        @Bean
        @StepScope
        public JdbcPagingItemReader<Order> partitionedOrderReader(
                DataSource dataSource,
                @Value("#{stepExecutionContext['minId']}") Long minId,
                @Value("#{stepExecutionContext['maxId']}") Long maxId) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause("order_id, customer_id, amount, status, ordered_at");
            provider.setFromClause("FROM orders");
            provider.setWhereClause(
                    "WHERE status = 'COMPLETED' AND order_id BETWEEN :minId AND :maxId");
            // 정렬 키는 페이징 리더의 생명입니다 — Step 06 참고.
            // 주의: Spring Batch 의 정렬 방향 enum 이름도 Order 라서 우리 도메인
            // Order 와 충돌합니다. import 하지 말고 완전한 이름으로 씁니다.
            provider.setSortKeys(Map.of(
                    "order_id", org.springframework.batch.item.database.Order.ASCENDING));

            return new JdbcPagingItemReaderBuilder<Order>()
                    .name("partitionedOrderReader")
                    .dataSource(dataSource)
                    .queryProvider(provider)
                    .parameterValues(Map.of("minId", minId, "maxId", maxId))
                    .pageSize(1000)
                    .rowMapper(new DataClassRowMapper<>(Order.class))
                    .build();
        }

        @Bean
        public Step settlementWorkerStep(JobRepository jobRepository,
                                         PlatformTransactionManager txManager,
                                         JdbcPagingItemReader<Order> partitionedOrderReader,
                                         DataSource dataSource) {
            return new StepBuilder("settlementWorkerStep", jobRepository)
                    .<Order, Settlement>chunk(1000, txManager)
                    .reader(partitionedOrderReader)
                    .processor(new SettlementProcessor())
                    .writer(buildSettlementWriter(dataSource))
                    .build();
        }
    }

    // =====================================================================
    // [13-9] gridSize 튜닝 — 스레드 풀과 커넥션 풀
    //
    // GRID_SIZES 를 돌며 같은 Job 을 반복 실행하는 러너입니다.
    // 전체가 약 2분 30초 걸리고, 끝나면 13-11 의 비교표와 같은 형태로
    // 콘솔에 요약을 찍습니다.
    //
    // 실행:
    //   ./gradlew bootRun -Pargs=--spring.batch.job.name=NONE,--grid.matrix=true
    // =====================================================================

    public static final int[] GRID_SIZES = {1, 4, 8, 16, 32};

    /**
     * 측정 결과 요약 출력.
     *
     * 기대 결과(13-11 과 동일한 경향):
     *   grid  1 →  62.4초   1.00배
     *   grid  4 →  17.2초   3.6배
     *   grid  8 →  11.2초   5.6배   ← 최적
     *   grid 16 →  12.9초   4.8배   ← 역전 시작 (커넥션 풀 20 포화)
     *   grid 32 →  24.6초   2.5배   ← 대기 지옥
     *
     * 병렬도는 단조 증가가 아닙니다. 최적점이 있습니다.
     */
    public static void printMatrix(Map<Integer, Long> elapsedMillisByGrid) {
        long baseline = elapsedMillisByGrid.getOrDefault(1, 62_400L);
        System.out.println("| gridSize | 소요 | 배수 |");
        System.out.println("|---|---|---|");
        for (int grid : GRID_SIZES) {
            Long ms = elapsedMillisByGrid.get(grid);
            if (ms == null) continue;
            System.out.printf("| %d | %.1f초 | %.2f배 |%n",
                    grid, ms / 1000.0, baseline / (double) ms);
        }
    }

    // =====================================================================
    // [13-10] 원격 청킹과 원격 파티셔닝 — 참고용 스케치
    //
    // ⚠️ 이 클래스는 컴파일은 되지만 실행되지 않습니다.
    //    RabbitMQ 와 spring-batch-integration 의존성이 없기 때문입니다.
    //    원격 구성의 뼈대를 읽기 위한 참고용입니다.
    //
    // 실행하려면 먼저 브로커를 띄우고 build.gradle 에 추가해야 합니다.
    //   implementation 'org.springframework.batch:spring-batch-integration'
    //   implementation 'org.springframework.boot:spring-boot-starter-amqp'
    // =====================================================================

    /*
    // @Configuration
    public static class RemoteSketch {

        // 원격 청킹 — 마스터가 "읽고", 워커가 "처리하고 쓴다".
        //   읽기는 여전히 마스터 혼자 하므로 읽기가 병목이면 효과 없음.
        //   processor 가 무거울 때(외부 API 호출 등) 유효합니다.
        //
        // 원격 파티셔닝 — 마스터가 "범위만 나눠 주고", 워커가 "읽고 처리하고 쓴다".
        //   읽기까지 분산되므로 로컬 파티셔닝의 자연스러운 확장입니다.
        //   대부분의 경우 원격 청킹보다 이쪽이 정답입니다.

        @Bean
        public MessageChannelPartitionHandler remotePartitionHandler(...) {
            MessageChannelPartitionHandler handler = new MessageChannelPartitionHandler();
            handler.setStepName("settlementWorkerStep");
            handler.setGridSize(8);
            handler.setMessagingOperations(messagingTemplate);
            return handler;
        }
    }
    */

    // =====================================================================
    // [참고] 이 스텝에서 쓴 검증 SQL 모음
    //
    // -- 누락 확인 (핵심). 70000 이 아니면 스레드 안전성 문제입니다.
    // SELECT COUNT(*) rows_written, COUNT(DISTINCT order_id) distinct_orders,
    //        SUM(net_amount) total_net FROM settlement;
    //
    // -- 메타데이터는 뭐라고 하는가 (70000/70000 이라고 거짓말합니다)
    // SELECT STEP_NAME, STATUS, READ_COUNT, WRITE_COUNT, COMMIT_COUNT, ROLLBACK_COUNT
    // FROM BATCH_STEP_EXECUTION ORDER BY STEP_EXECUTION_ID DESC LIMIT 1;
    //
    // -- 파티션별 처리 건수 (스큐 확인)
    // SELECT STEP_NAME, READ_COUNT, WRITE_COUNT
    // FROM BATCH_STEP_EXECUTION
    // WHERE STEP_NAME LIKE 'settlementWorkerStep:partition%'
    // ORDER BY STEP_NAME;
    //
    // -- 어느 주문이 누락됐나
    // SELECT o.order_id FROM orders o
    // LEFT JOIN settlement s ON s.order_id = o.order_id
    // WHERE o.status = 'COMPLETED' AND s.order_id IS NULL
    // LIMIT 20;
    //
    // -- 커넥션 풀 상태 (logging.level.com.zaxxer.hikari=DEBUG 필요)
    // -- HikariPool - Pool stats (total=20, active=16, idle=4, waiting=12)
    // =====================================================================
}

package com.example.batch.step13;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.jdbc.core.JdbcTemplate;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;

/**
 * Step 13 — 연습문제 (6문제)
 *
 * 정답은 Solution.java. 먼저 직접 풀어 보십시오.
 *
 * ─────────────────────────────────────────────────────────────────────────
 * 공통 준비
 *
 * 모든 측정 전에 반드시 초기화하십시오. 빼먹으면 이전 실행 결과가 남아
 * 누락을 발견할 수 없습니다.
 *
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 *
 * 검증은 항상 이 쿼리로 합니다. distinct_orders 가 70000 이어야 정상입니다.
 *
 *   SELECT COUNT(*) rows_written, COUNT(DISTINCT order_id) distinct_orders
 *   FROM settlement;
 * ─────────────────────────────────────────────────────────────────────────
 */
public class Exercise {

    // =====================================================================
    // 문제 1. 스레드 안전하지 않은 Reader 를 두 가지 방법으로 고치기
    //
    // 아래 BrokenListReader 는 13-3 의 NaiveOrderReader 와 같은 결함을
    // 가지고 있습니다.
    //
    // (a) 이 Reader 가 왜 스레드 안전하지 않은지 한 문장으로 설명하십시오.
    // (b) 두 가지 방법으로 고치십시오.
    //     ① Reader 내부를 고쳐서 (힌트: AtomicInteger)
    //     ② Reader 는 그대로 두고 바깥에서 감싸서 (힌트: Step 13-4)
    // (c) 두 방법의 성능을 각각 측정해 비교하십시오. 어느 쪽이 빠릅니까?
    //     그리고 그 차이가 나는 이유는 무엇입니까?
    // =====================================================================

    public static class BrokenListReader implements ItemReader<Order> {

        private final List<Order> orders;
        private int index = 0;

        public BrokenListReader(List<Order> orders) {
            this.orders = orders;
        }

        @Override
        public Order read() {
            if (index >= orders.size()) {
                return null;
            }
            return orders.get(index++);
        }
    }

    // (a) 왜 스레드 안전하지 않은가?
    // 여기에 작성:
    //

    /** ① AtomicInteger 로 인덱스를 원자화한 버전. */
    public static class FixedByAtomicReader implements ItemReader<Order> {

        private final List<Order> orders;
        // 여기에 작성: 인덱스 필드를 원자적으로 바꾸십시오
        //

        public FixedByAtomicReader(List<Order> orders) {
            this.orders = orders;
        }

        @Override
        public Order read() {
            // 여기에 작성:
            //
            return null;
        }
    }

    // ② SynchronizedItemStreamReader 로 감싸는 방법
    // 여기에 작성: (빈 정의 형태로)
    //

    // (c) 측정 결과
    //     ① AtomicInteger  : ____초
    //     ② Synchronized   : ____초
    //     더 빠른 쪽과 그 이유:
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 2. SynchronizedItemStreamReader 로 감싸도 여전히 틀리는 경우
    //
    // ⚠️ 이 스텝에서 가장 어려운 문제입니다.
    //
    // 아래 RunningTotalProcessor 를 문제 1 에서 고친 Reader 와 함께
    // 멀티스레드 Step(스레드 4개)에 넣고 돌리십시오.
    //
    // (a) Reader 를 완벽히 동기화했는데도 runningTotal 이 매 실행마다
    //     다르게 나옵니다. 왜입니까?
    // (b) settlement 테이블의 행 수는 70,000 이 맞습니까? 그렇다면
    //     "무엇이" 틀린 것입니까?
    // (c) 이 Processor 를 올바르게 고치십시오. 힌트: 두 갈래의 답이 있습니다.
    //     하나는 "누적을 스레드 안전하게 만드는" 것이고,
    //     다른 하나는 "애초에 누적하지 않는" 것입니다. 어느 쪽이 정답입니까?
    // =====================================================================

    public static class RunningTotalProcessor implements ItemProcessor<Order, Settlement> {

        private BigDecimal runningTotal = BigDecimal.ZERO;   // ← 공유 가변 상태

        @Override
        public Settlement process(Order order) {
            runningTotal = runningTotal.add(order.amount());
            // ... Settlement 변환 (Practice 의 SettlementProcessor 와 동일)
            return null;
        }

        public BigDecimal getRunningTotal() {
            return runningTotal;
        }
    }

    // (a) 왜 매번 다른가?
    // 여기에 작성:
    //

    // (b) settlement 행 수는 맞는가? 그렇다면 무엇이 틀렸는가?
    // 여기에 작성:
    //

    // (c) 올바른 수정
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 3. customer_id 기준 Partitioner 작성하기 (등급 스큐 고려)
    //
    // order_id 대신 customer_id 로 나누는 Partitioner 를 작성하십시오.
    //
    // 참고 — 시드 데이터의 규칙:
    //   customer_id = ((order_id - 1) % 1000) + 1     (고객 1,000명)
    //   등급        = customer_id % 4                  (BRONZE/SILVER/GOLD/VIP)
    //   따라서 고객 한 명당 주문은 정확히 100건, 그중 COMPLETED 는 70건.
    //
    // (a) customer_id 를 gridSize 개의 "연속 범위"로 나누는 Partitioner 를
    //     작성하십시오. (1~250, 251~500, ...)
    // (b) gridSize=4 로 돌린 뒤, 파티션별 등급 분포를 아래 쿼리로 확인하십시오.
    //
    //     SELECT s.customer_id % 4 AS grade_key, COUNT(*)
    //     FROM settlement s WHERE s.customer_id BETWEEN 1 AND 250
    //     GROUP BY grade_key;
    //
    //     등급이 파티션마다 고르게 섞여 있습니까? 아니라면 왜입니까?
    // (c) 등급이 고르게 섞이도록 분할 방식을 바꾸십시오.
    //     힌트: 범위(range) 분할 대신 해시(hash) 분할.
    // (d) 범위 분할과 해시 분할은 각각 어떤 상황에 적합합니까?
    // =====================================================================

    public static class CustomerIdPartitioner implements Partitioner {

        private final JdbcTemplate jdbcTemplate;

        public CustomerIdPartitioner(DataSource dataSource) {
            this.jdbcTemplate = new JdbcTemplate(dataSource);
        }

        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            // 여기에 작성: (a) 연속 범위 분할
            //
            return null;
        }
    }

    // (b) 등급 분포 확인 결과와 그 이유
    // 여기에 작성:
    //

    // (c) 해시 분할 버전
    // 여기에 작성:
    //

    // (d) 범위 분할 vs 해시 분할의 적합한 상황
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 4. @StepScope 를 빼면 어떤 예외가 어느 시점에 나는가
    //
    // Practice.java 의 Partitioning.partitionedOrderReader 에서
    // @StepScope 한 줄만 주석 처리하고 실행하십시오.
    //
    // (a) 예외를 먼저 "예측"해 적으십시오. 실행하기 전에 적어야 의미가 있습니다.
    //     - 예외 클래스명:
    //     - 발생 시점: (애플리케이션 기동 중 / Job 실행 중 / 청크 처리 중)
    // (b) 실제로 실행해 확인하고, 예측과 다르면 왜 틀렸는지 적으십시오.
    // (c) 이 실패는 "시끄러운 실패"입니까 "조용한 실패"입니까?
    //     그것이 왜 다행인지 13-8 의 함정 블록과 연결해 설명하십시오.
    // =====================================================================

    // (a) 예측
    // 여기에 작성:
    //

    // (b) 실제 결과
    // 여기에 작성:
    //

    // (c) 시끄러운 실패 vs 조용한 실패
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 5. 커넥션 풀 20 에서 안전한 최대 gridSize 계산하기
    //
    // application.yml 의 설정은 maximum-pool-size: 20 입니다.
    //
    // (a) 동시에 도는 워커 하나가 커넥션을 몇 개나 필요로 합니까?
    //     (힌트: Reader 가 하나, 그리고 또 하나가 더 필요합니다. 무엇입니까?)
    // (b) 마스터 Step 과 Partitioner 자신도 커넥션을 씁니다. 몇 개를
    //     여유로 남겨야 합니까?
    // (c) 안전한 최대 동시 워커 수를 계산하고 공식으로 정리하십시오.
    // (d) 계산한 값을 13-11 의 실측표와 대조하십시오. 맞습니까?
    //
    // ⚠️ 답만 쓰지 마십시오. 근거를 실제 로그로 만들어야 합니다.
    //    application.yml 에 아래를 추가하면 30초마다 풀 상태가 찍힙니다.
    //      logging.level.com.zaxxer.hikari: DEBUG
    //    다음과 같은 줄을 찾으십시오.
    //      HikariPool - Pool stats (total=20, active=16, idle=4, waiting=12)
    //    waiting 이 0 보다 크면 이미 포화된 것입니다.
    // =====================================================================

    // (a) 워커 하나당 커넥션 수
    // 여기에 작성:
    //

    // (b) 여유분
    // 여기에 작성:
    //

    // (c) 공식과 계산
    // 여기에 작성:
    //

    // (d) 실측표와의 대조
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 6. 멀티스레드 Step 과 파티셔닝 중 무엇을 쓸 것인가 (4개 시나리오)
    //
    // 각 시나리오에 대해 ① 어느 방식을 쓸지 ② 그 이유 ③ 주의점을 적으십시오.
    //
    // 시나리오 A ─ 100만 행을 읽어 단순 계산 후 다른 테이블에 쓴다.
    //   Reader 는 JdbcPagingItemReader. 재시작이 반드시 되어야 한다.
    //
    // 시나리오 B ─ 5만 건 각각에 대해 외부 결제사 API 를 호출한다.
    //   API 응답이 건당 평균 200ms 걸린다. CPU 는 거의 안 쓴다.
    //
    // 시나리오 C ─ 하나의 거대한 CSV 파일(2GB)을 읽어 DB 에 적재한다.
    //   FlatFileItemReader 를 쓴다.
    //
    // 시나리오 D ─ 정산 배치를 8대의 서버에 나눠서 돌리고 싶다.
    //   단일 서버로는 시간 안에 끝나지 않는다.
    // =====================================================================

    // 시나리오 A
    // 여기에 작성:
    //

    // 시나리오 B
    // 여기에 작성:
    //

    // 시나리오 C
    // 여기에 작성:
    //

    // 시나리오 D
    // 여기에 작성:
    //
}

package com.example.batch.step13;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.support.SynchronizedItemStreamReader;
import org.springframework.jdbc.core.JdbcTemplate;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Step 13 — 연습문제 정답과 해설
 *
 * 문제를 직접 풀어 본 뒤에 여십시오.
 */
public class Solution {

    // =====================================================================
    // 정답 1. 스레드 안전하지 않은 Reader 를 두 가지 방법으로 고치기
    // =====================================================================

    /*
     * (a) 왜 스레드 안전하지 않은가
     *
     *   `orders.get(index++)` 의 index++ 가 원자적이지 않기 때문입니다.
     *
     *   index++ 는 바이트코드에서 세 단계로 나뉩니다.
     *     ① getfield index      (읽기)
     *     ② iadd 1              (더하기)
     *     ③ putfield index      (쓰기)
     *
     *   스레드 A 와 B 가 ①을 동시에 실행하면 둘 다 같은 값(예: 500)을 읽습니다.
     *   둘 다 500번 아이템을 반환하고, 둘 다 index 를 501 로 씁니다.
     *   결과적으로
     *     - 500번 아이템이 두 번 처리되고
     *     - 501번이 되었어야 할 한 자리가 통째로 사라집니다.
     *
     *   settlement.order_id 에 UNIQUE 가 걸려 있고 Writer 가 멱등이라
     *   "중복"은 흡수되어 안 보이고, "누락"만 남습니다.
     *   그래서 70,000 이 아니라 69,213 이 되는 것입니다.
     *
     *   ⚠️ 이 결함의 가장 고약한 점은 결정론적이지 않다는 것입니다.
     *      스레드 스케줄링에 의존하므로 실행할 때마다 손실 건수가 다릅니다.
     *      69213 / 68940 / 69551 ...
     *
     *      역으로, 이것이 진단 기법이 됩니다.
     *      **입력이 완전히 결정론적인데 출력이 매번 다르면, 원인은 거의 항상
     *      스레드 안전성입니다.** 프로젝트 셋업에서 RAND() 를 한 번도 쓰지
     *      않고 나머지 연산만으로 시드를 만든 이유가 바로 이것입니다.
     */

    /** ① AtomicInteger 로 인덱스를 원자화 — 빠른 특수해. */
    public static class FixedByAtomicReader implements ItemReader<Order> {

        private final List<Order> orders;
        private final AtomicInteger index = new AtomicInteger(0);

        public FixedByAtomicReader(List<Order> orders) {
            this.orders = orders;
        }

        @Override
        public Order read() {
            // getAndIncrement() 는 CAS(Compare-And-Swap) 기반이라
            // 읽기+증가+쓰기가 하나의 원자적 연산으로 처리됩니다.
            int i = index.getAndIncrement();
            if (i >= orders.size()) {
                return null;
            }
            return orders.get(i);
        }
    }

    /*
     * ② SynchronizedItemStreamReader 로 감싸기 — 느린 일반해.
     *
     *   @Bean
     *   public SynchronizedItemStreamReader<Order> safeReader(DataSource ds) {
     *       SynchronizedItemStreamReader<Order> reader = new SynchronizedItemStreamReader<>();
     *       reader.setDelegate(orderCursorReader(ds));
     *       return reader;
     *   }
     *
     *   read() 전체를 synchronized 로 감쌉니다.
     *
     * (c) 측정 결과와 해석
     *
     *   ① AtomicInteger  : 18.4초
     *   ② Synchronized   : 21.6초
     *
     *   AtomicInteger 가 약 17% 빠릅니다. 이유는 두 가지입니다.
     *
     *   - CAS 는 락을 잡지 않습니다. 경합이 없으면 사실상 공짜이고,
     *     경합이 있어도 커널 레벨의 컨텍스트 스위칭이 없습니다.
     *     synchronized 는 경합 시 스레드를 블로킹시킵니다.
     *   - 임계 구역의 크기가 다릅니다. AtomicInteger 버전은 "인덱스 증가"
     *     한 순간만 원자적이면 되고, 실제 `orders.get(i)` 는 여러 스레드가
     *     동시에 할 수 있습니다. Synchronized 버전은 read() 호출 전체를
     *     직렬화하므로 delegate 의 DB 접근까지 한 줄로 세웁니다.
     *
     *   ⚠️ 그러나 AtomicInteger 해법에는 중요한 전제가 있습니다.
     *      **`orders.get(i)` 자체가 스레드 안전해야** 합니다.
     *      여기서는 불변 List 를 읽기만 하므로 안전합니다. 하지만
     *      JdbcCursorItemReader 처럼 내부에 ResultSet 커서를 들고 있는
     *      Reader 에는 이 방법을 쓸 수 없습니다. 인덱스를 원자화해 봐야
     *      ResultSet.next() 자체가 스레드 안전하지 않기 때문입니다.
     *
     *   정리하면 전형적인 "빠른 특수해 vs 느린 일반해" 트레이드오프입니다.
     *
     *     | | AtomicInteger | SynchronizedItemStreamReader |
     *     |---|---|---|
     *     | 속도 | 18.4초 | 21.6초 |
     *     | 적용 범위 | 내가 만든 Reader, 백킹 자료구조가 안전할 때만 | 모든 ItemStreamReader |
     *     | 구현 난이도 | Reader 를 직접 고쳐야 함 | 한 줄로 감싸기 |
     *
     *   실무에서는 대개 ②를 씁니다. 3.2초를 아끼려고 Reader 내부 구현에
     *   대한 가정을 코드에 심는 것은 대체로 손해입니다.
     *   그리고 애초에 둘 다 파티셔닝(11.2초)보다 느립니다 — 정답 6 참고.
     */

    // =====================================================================
    // 정답 2. Synchronized 로 감싸도 여전히 틀리는 경우
    //
    // 이 문제가 이 스텝의 진짜 교훈입니다.
    // =====================================================================

    /*
     * (a) 왜 매번 다른가
     *
     *   SynchronizedItemStreamReader 는 이름 그대로 **Reader 만** 보호합니다.
     *   Processor 는 전혀 건드리지 않습니다.
     *
     *   RunningTotalProcessor.process() 안의
     *     runningTotal = runningTotal.add(order.amount());
     *   는 읽기 → 계산 → 쓰기의 3단계이고, index++ 와 정확히 같은 결함을
     *   가지고 있습니다. BigDecimal 이 불변 객체라는 사실은 아무 도움이
     *   되지 않습니다. 불변인 것은 BigDecimal 이지 `runningTotal` 필드가
     *   아니기 때문입니다.
     *
     *   4개 스레드가 동시에 더하면 갱신이 서로를 덮어써서 합계가 실제보다
     *   작게 나옵니다. 그리고 얼마나 작을지는 스케줄링에 달렸습니다.
     *
     * (b) settlement 행 수는 맞는가?
     *
     *   맞습니다. 정확히 70,000 입니다.
     *
     *   **이것이 이 문제의 핵심입니다.** Reader 를 동기화했으므로 모든
     *   아이템이 정확히 한 번씩 읽히고, 처리되고, 쓰였습니다. 저장된
     *   데이터는 완벽합니다.
     *
     *   틀린 것은 **Processor 가 곁다리로 들고 있던 집계값**입니다.
     *   그리고 이 값은 DB 에 없고 메모리에만 있어서, 검증 쿼리로는
     *   절대 잡히지 않습니다. 로그로 찍거나 리포트에 넣는 순간
     *   조용히 틀린 숫자가 됩니다.
     *
     *   "Reader 만 고치면 된다"는 오해가 여기서 깨집니다.
     *   스레드 안전성은 Reader·Processor·Writer·Listener **전부**의 문제입니다.
     *
     * (c) 올바른 수정 — 두 갈래 중 무엇이 정답인가
     *
     *   갈래 ① 누적을 스레드 안전하게 만든다
     *
     *     private final LongAdder totalCents = new LongAdder();
     *     ...
     *     totalCents.add(order.amount().movePointRight(2).longValueExact());
     *
     *     LongAdder 는 경합 시 내부 셀로 분산 누적해서 AtomicLong 보다
     *     빠릅니다. (BigDecimal 은 원자적 누적 API 가 없으므로 정수 센트로
     *     바꿔서 다룹니다.)
     *
     *   갈래 ② 애초에 누적하지 않는다  ← **이쪽이 정답입니다**
     *
     *     ItemProcessor 는 **순수 함수여야 합니다.** 입력 하나를 받아
     *     출력 하나를 내놓을 뿐, 상태를 남기지 않아야 합니다.
     *     이것은 멀티스레드 때문에 생긴 규칙이 아니라 원래의 계약입니다.
     *
     *     집계가 필요하면 집계를 담당하는 곳에서 하십시오.
     *
     *     - Step 이 끝난 뒤의 합계가 필요하다면
     *       → StepExecutionListener.afterStep() 에서 SQL 로 집계
     *         SELECT SUM(net_amount) FROM settlement WHERE ...
     *       DB 가 이미 정확한 값을 갖고 있는데 애플리케이션 메모리에서
     *       따로 세는 것은 중복이자 버그의 원천입니다.
     *
     *     - 건수만 필요하다면
     *       → StepExecution.getWriteCount() 를 쓰십시오.
     *         프레임워크가 이미 스레드 안전하게 세고 있습니다.
     *
     *   갈래 ①은 "동작하게" 만들고, 갈래 ②는 "애초에 틀릴 수 없게" 만듭니다.
     *   동시성 문제는 언제나 후자가 정답입니다. 공유 상태를 안전하게
     *   다루는 것보다 공유 상태를 없애는 것이 쉽습니다.
     */

    /** 상태를 갖지 않는 올바른 Processor. Practice 의 SettlementProcessor 와 동일한 형태. */
    public static class StatelessProcessor implements ItemProcessor<Order, Settlement> {
        @Override
        public Settlement process(Order order) {
            // 필드가 없습니다. 그래서 스레드 안전성을 고민할 필요 자체가 없습니다.
            return null;    // 실제 변환은 Practice.SettlementProcessor 참고
        }
    }

    // =====================================================================
    // 정답 3. customer_id 기준 Partitioner
    // =====================================================================

    /** (a) 연속 범위 분할 — 등급 스큐가 생깁니다. */
    public static class CustomerRangePartitioner implements Partitioner {

        private final JdbcTemplate jdbcTemplate;

        public CustomerRangePartitioner(DataSource dataSource) {
            this.jdbcTemplate = new JdbcTemplate(dataSource);
        }

        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            Integer min = jdbcTemplate.queryForObject(
                    "SELECT MIN(customer_id) FROM customers", Integer.class);
            Integer max = jdbcTemplate.queryForObject(
                    "SELECT MAX(customer_id) FROM customers", Integer.class);

            int targetSize = (max - min) / gridSize + 1;
            Map<String, ExecutionContext> result = new LinkedHashMap<>();
            int start = min;
            for (int i = 0; i < gridSize; i++) {
                ExecutionContext ctx = new ExecutionContext();
                int end = Math.min(start + targetSize - 1, max);
                ctx.putInt("minCustomerId", start);
                ctx.putInt("maxCustomerId", end);
                result.put("partition" + i, ctx);
                start = end + 1;
            }
            return result;
        }
    }

    /*
     * (b) 등급 분포 확인 결과
     *
     *   gridSize=4 로 나누면 파티션은 1~250, 251~500, 501~750, 751~1000 입니다.
     *
     *   SELECT customer_id % 4 AS grade_key, COUNT(*) FROM customers
     *   WHERE customer_id BETWEEN 1 AND 250 GROUP BY grade_key;
     *
     *   +-----------+----------+
     *   | grade_key | COUNT(*) |
     *   +-----------+----------+
     *   |         0 |       62 |
     *   |         1 |       63 |
     *   |         2 |       63 |
     *   |         3 |       62 |
     *   +-----------+----------+
     *
     *   ... 사실 고르게 나옵니다.
     *
     *   왜냐하면 등급이 `customer_id % 4` 로 정해지는데, 250 은 4의 배수에
     *   가까워서 어떤 연속 구간을 잡아도 나머지 0/1/2/3 이 거의 같은 수로
     *   섞이기 때문입니다.
     *
     *   ⚠️ 그러니까 이 문제의 진짜 답은 이것입니다:
     *      **우리 실습 데이터에서는 스큐가 안 생깁니다.**
     *      시드를 나머지 연산으로 만들었기 때문에, 어떤 방식으로 잘라도
     *      균등합니다. 이건 실습 데이터의 성질이지 일반적인 성질이 아닙니다.
     *
     *   실제 데이터였다면 이렇게 됩니다.
     *     - VIP 고객은 보통 초기 가입자라 customer_id 가 작은 쪽에 몰립니다.
     *     - 그런데 VIP 는 주문 건수도 많습니다.
     *     - 따라서 파티션 0 (customer_id 1~250) 이 다른 파티션의 3~5배
     *       분량을 떠안게 됩니다.
     *     - 8개 파티션 중 7개가 2초에 끝나고 하나가 40초를 도는 상황.
     *       전체 소요는 40초입니다. 병렬화의 이득이 거의 사라집니다.
     *
     *   파티션별 실제 처리량은 이 쿼리로 확인합니다.
     *
     *     SELECT STEP_NAME, READ_COUNT, WRITE_COUNT,
     *            TIMESTAMPDIFF(SECOND, START_TIME, END_TIME) AS secs
     *     FROM BATCH_STEP_EXECUTION
     *     WHERE STEP_NAME LIKE 'settlementWorkerStep:partition%'
     *     ORDER BY secs DESC;
     *
     *   READ_COUNT 가 파티션마다 2배 이상 차이 나면 스큐입니다.
     */

    /** (c) 해시 분할 — 어떤 분포에서도 균등합니다. */
    public static class CustomerHashPartitioner implements Partitioner {

        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            Map<String, ExecutionContext> result = new LinkedHashMap<>();
            for (int i = 0; i < gridSize; i++) {
                ExecutionContext ctx = new ExecutionContext();
                // WHERE customer_id % :gridSize = :remainder
                ctx.putInt("gridSize", gridSize);
                ctx.putInt("remainder", i);
                result.put("partition" + i, ctx);
            }
            return result;
        }
    }

    /*
     *   해시 분할 Reader 의 WHERE 절:
     *     WHERE status = 'COMPLETED' AND customer_id % :gridSize = :remainder
     *
     *   ⚠️ 성능상의 큰 함정이 하나 있습니다.
     *      `customer_id % 4 = 0` 은 컬럼에 함수를 씌운 형태라 **인덱스를
     *      타지 못합니다.** 각 파티션이 풀스캔을 하게 되어, 8개 파티션이면
     *      테이블을 8번 풀스캔합니다. 범위 분할보다 오히려 느려질 수 있습니다.
     *
     *      실무에서는 이 때문에 샤드 키 컬럼을 아예 따로 둡니다.
     *        ALTER TABLE orders ADD COLUMN shard_key TINYINT
     *          GENERATED ALWAYS AS (customer_id % 16) STORED,
     *          ADD INDEX idx_shard (shard_key, status);
     *      생성 컬럼에 인덱스를 걸면 해시 분할도 인덱스를 탑니다.
     *
     * (d) 범위 분할 vs 해시 분할
     *
     *   | | 범위 분할 | 해시 분할 |
     *   |---|---|---|
     *   | 인덱스 | PK 범위라 잘 탐 | 함수 조건이라 못 탐 (생성 컬럼 필요) |
     *   | 스큐 | 데이터 분포에 취약 | 항상 균등 |
     *   | 신규 데이터 | 마지막 파티션에 몰림 | 고르게 분산 |
     *   | 재실행 일관성 | gridSize 바뀌면 경계 이동 | gridSize 바뀌면 전체 재배치 |
     *
     *   실무 지침:
     *   - **기본은 범위 분할.** 인덱스를 타는 것이 압도적으로 중요합니다.
     *   - 스큐가 확인되면 → gridSize 를 크게(100) 두고 풀은 작게(8).
     *     조각이 잘게 쪼개지면 큰 조각 하나가 전체를 지배하는 일이 줄고,
     *     빈 스레드가 다음 조각을 바로 집어 갑니다. 13-9 의 #11 구성입니다.
     *   - 그래도 안 되면 → 샤드 키 생성 컬럼 + 해시 분할.
     */

    // =====================================================================
    // 정답 4. @StepScope 를 빼면
    // =====================================================================

    /*
     * (a)(b) 예외와 시점
     *
     *   org.springframework.expression.spel.SpelEvaluationException:
     *     EL1007E: Property or field 'minId' cannot be found on null
     *
     *   발생 시점은 **애플리케이션 기동 중**입니다. Job 실행 중이 아닙니다.
     *
     *   실제 로그:
     *
     *   ERROR 44821 --- [  main] o.s.boot.SpringApplication : Application run failed
     *   org.springframework.beans.factory.UnsatisfiedDependencyException:
     *     Error creating bean with name 'partitionedOrderReader' defined in ...
     *     Failed to convert value of type 'null' ...
     *   Caused by: org.springframework.expression.spel.SpelEvaluationException:
     *     EL1007E: Property or field 'minId' cannot be found on null
     *
     *   이유는 단순합니다. @StepScope 가 없으면 이 @Bean 메서드는 싱글턴이라
     *   **컨텍스트 기동 시점에 딱 한 번** 호출됩니다. 그 시점에는 Job 이
     *   시작도 안 했으므로 stepExecutionContext 자체가 존재하지 않습니다.
     *   SpEL 이 null 위에서 'minId' 를 찾으려다 터집니다.
     *
     *   @StepScope 를 붙이면 스코프 프록시가 끼어들어, 실제 빈 생성이
     *   **각 Step 실행 시점으로 미뤄집니다.** 그때는 파티션이 만들어져
     *   stepExecutionContext 에 minId/maxId 가 들어 있습니다.
     *   이것이 "늦은 바인딩(late binding)"입니다.
     *
     * (c) 시끄러운 실패 vs 조용한 실패
     *
     *   이것은 **시끄러운 실패**이고, 그래서 다행입니다.
     *
     *   - 애플리케이션이 아예 뜨지 않습니다.
     *   - 배포 파이프라인에서 즉시 걸립니다.
     *   - 운영에 나갈 수가 없습니다.
     *
     *   13-8 의 함정 블록이 지적한 반대 사례와 대조해 보십시오.
     *   파라미터를 SpEL 대신 `new HashMap<>()` 같은 빈 값으로 넘기도록
     *   잘못 짜면, 예외 없이 `WHERE order_id BETWEEN null AND null` 이
     *   되어 **0건을 조회하고 Job 은 COMPLETED 로 끝납니다.**
     *   settlement 는 비어 있는데 배치는 성공했다고 보고합니다.
     *
     *   같은 실수(늦은 바인딩 실패)인데 한쪽은 기동조차 안 되고 한쪽은
     *   조용히 아무것도 안 합니다. 후자가 비교할 수 없이 위험합니다.
     *
     *   이 코스가 반복해서 말하는 것이 이것입니다:
     *   **문법 에러는 금방 고치지만, 에러 없이 조용히 틀리는 코드가
     *   진짜 위험합니다.**
     */

    // =====================================================================
    // 정답 5. 커넥션 풀 20 에서 안전한 최대 gridSize
    // =====================================================================

    /*
     * (a) 워커 하나당 커넥션 수 = 2
     *
     *   ① Reader 의 커넥션 — JdbcPagingItemReader 가 페이지를 조회할 때
     *   ② 청크 트랜잭션의 커넥션 — Writer 의 INSERT 와 JobRepository 의
     *      메타데이터 갱신(BATCH_STEP_EXECUTION UPDATE)이 같은 트랜잭션에
     *      묶여 하나를 더 씁니다.
     *
     *   ②를 놓치기 쉽습니다. "Reader 하나니까 워커당 1개"라고 계산하면
     *   실제의 절반으로 잡게 되어 반드시 풀이 마릅니다.
     *
     *   ⚠️ 커서 기반 Reader 는 더 나쁩니다. JdbcCursorItemReader 는
     *      Step 이 끝날 때까지 커넥션을 **계속 붙들고 있습니다.**
     *      페이징은 페이지마다 반납하지만 커서는 반납하지 않습니다.
     *      파티셔닝에 페이징 Reader 를 쓰는 이유 중 하나입니다.
     *
     * (b) 여유분 = 4
     *
     *   - 마스터 Step 자신 : 1
     *   - Partitioner 의 MIN/MAX 초기 쿼리 : 1
     *   - JobRepository 의 Job 레벨 메타데이터 갱신 여유 : 2
     *
     * (c) 공식
     *
     *   최대 동시 워커 수 = (maximum-pool-size - 여유분) / 워커당 커넥션 수
     *                    = (20 - 4) / 2
     *                    = 8
     *
     *   따라서 **동시 워커 8개가 상한**입니다.
     *
     *   여기서 gridSize 와 풀 크기를 구분해야 합니다.
     *     - 동시에 도는 워커 수 = TaskExecutor 의 corePoolSize → 8 이 상한
     *     - gridSize = 조각 수 → 8 보다 커도 됩니다. 순차로 처리될 뿐입니다.
     *
     * (d) 실측표와의 대조 — 정확히 맞습니다
     *
     *   #8  파티셔닝 grid 8  → 11.2초  5.6배  ← 최적
     *   #9  파티셔닝 grid 16 → 12.9초  4.8배  ← 역전
     *   #10 파티셔닝 grid 32 → 24.6초  2.5배  ← 대기 지옥
     *
     *   계산상의 상한 8 에서 최적점이 나오고, 그 이상에서 성능이 떨어집니다.
     *   grid 16 일 때 Hikari 로그를 보면 근거가 보입니다.
     *
     *     HikariPool - Pool stats (total=20, active=20, idle=0, waiting=12)
     *
     *   active=20 으로 풀이 완전히 소진됐고 12개 요청이 대기 중입니다.
     *   워커들이 일을 하는 게 아니라 커넥션을 기다리며 서로를 막고 있습니다.
     *   grid 32 에서는 이 대기가 심해져 단일 스레드의 2.5배까지 떨어집니다.
     *
     *   ⚠️ 최악의 경우는 성능 저하가 아니라 데드락입니다.
     *      워커가 Reader 용 커넥션 1개를 잡은 채 트랜잭션용 2번째 커넥션을
     *      기다리는데, 모든 워커가 같은 상태면 아무도 진행하지 못합니다.
     *      결국 이렇게 죽습니다.
     *        HikariPool - Connection is not available,
     *        request timed out after 30000ms
     *      "워커당 2개"를 정확히 계산해야 하는 진짜 이유입니다.
     *
     *   #11 파티셔닝 grid 100 / 풀 8 → 11.9초 5.2배
     *
     *   이 구성이 실무에서 권장됩니다. 최적(11.2초)과 거의 같으면서
     *   스큐에 훨씬 강합니다. 조각이 잘아서 한 조각이 오래 걸려도 전체를
     *   지배하지 않고, 빈 스레드가 즉시 다음 조각을 집어 갑니다.
     *   **gridSize 는 크게, 풀은 계산된 상한으로.**
     */

    // =====================================================================
    // 정답 6. 4개 시나리오 판정
    //
    // 이 표가 실무에서 그대로 쓸 수 있는 의사결정 체크리스트입니다.
    // =====================================================================

    /*
     * 시나리오 A ─ 100만 행, 단순 계산, 재시작 필수
     *
     *   → **파티셔닝**
     *
     *   재시작이 반드시 되어야 한다는 조건에서 이미 멀티스레드 Step 은
     *   탈락입니다. 멀티스레드 Step 은 saveState(false) 가 사실상 필수라
     *   재시작을 포기해야 하기 때문입니다(13-5).
     *
     *   파티셔닝은 각 워커 Step 이 독립된 StepExecution 을 가지므로
     *   ExecutionContext 가 정상적으로 저장되고, 실패한 파티션만 골라
     *   재시작됩니다. 이미 COMPLETED 인 파티션은 건너뜁니다.
     *
     *   주의점: PK 범위로 나누되 스큐를 확인할 것. gridSize 는 크게(100),
     *   풀은 (pool-4)/2 로.
     *
     *
     * 시나리오 B ─ 5만 건, 건당 외부 API 200ms, CPU 거의 안 씀
     *
     *   → **멀티스레드 Step**  ← 멀티스레드가 정답인 유일한 시나리오
     *
     *   이 경우가 특별한 이유는 **병목이 DB 도 CPU 도 아니라 네트워크
     *   대기**이기 때문입니다.
     *
     *   단일 스레드면 5만 건 × 200ms = 약 2시간 47분입니다. 스레드가
     *   대기하는 동안 아무 일도 안 합니다. 스레드를 32개로 늘리면 한
     *   스레드가 응답을 기다리는 동안 다른 31개가 각자 요청을 날립니다.
     *   이론상 32배, 약 5분입니다.
     *
     *   여기서는 커넥션 풀 계산이 병목이 아닙니다. 워커들이 대부분의
     *   시간을 DB 밖(HTTP 대기)에서 보내기 때문에, 스레드 수를 커넥션
     *   풀보다 훨씬 크게 잡아도 됩니다.
     *
     *   주의점 세 가지:
     *   - Reader 를 반드시 동기화하거나 스레드 안전한 것으로 쓸 것.
     *   - 재시작이 안 되므로, 실패 시 처음부터 다시 돌 것을 감수하거나
     *     Writer 를 멱등으로 만들어 중복 호출을 흡수할 것.
     *   - **상대방 API 의 rate limit 를 반드시 확인할 것.** 32 스레드로
     *     때리다 차단당하면 배치가 아니라 사고입니다.
     *
     *
     * 시나리오 C ─ 2GB CSV 하나, FlatFileItemReader
     *
     *   → **둘 다 아님. 단일 스레드로 두거나, 파일을 먼저 쪼갤 것.**
     *
     *   FlatFileItemReader 는 파일 포인터라는 본질적인 순차 상태를
     *   가집니다. 여러 스레드가 하나의 파일 스트림을 공유하면 줄이
     *   섞이거나 잘립니다. Synchronized 로 감싸면 정확해지지만 읽기가
     *   완전히 직렬화되어 이득이 거의 없습니다 — 그리고 이 시나리오에서는
     *   읽기가 바로 병목입니다.
     *
     *   제대로 하려면 파일을 물리적으로 나눕니다.
     *     split -l 250000 big.csv part_
     *   그리고 **파일 하나당 파티션 하나**를 배정합니다. MultiResourcePartitioner
     *   가 정확히 이 용도의 기본 제공 Partitioner 입니다.
     *
     *   교훈: 나눌 수 없는 자원(단일 파일 스트림)을 억지로 병렬화하지 말고,
     *   자원 자체를 나눌 것.
     *
     *
     * 시나리오 D ─ 8대 서버에 분산
     *
     *   → **원격 파티셔닝** (원격 청킹 아님)
     *
     *   여러 JVM 으로 나가는 순간 TaskExecutorPartitionHandler 는 쓸 수
     *   없습니다. MessageChannelPartitionHandler 로 바꾸고 RabbitMQ 나
     *   Kafka 로 파티션 정보를 워커 서버들에 보냅니다.
     *
     *   원격 청킹이 아니라 원격 파티셔닝인 이유:
     *   - 원격 청킹은 **마스터가 혼자 읽어서** 워커에 아이템을 뿌립니다.
     *     읽기가 분산되지 않으므로, 읽기가 병목이면 8대를 붙여도 소용없고
     *     오히려 직렬화 비용만 늘어납니다.
     *   - 원격 파티셔닝은 마스터가 **범위만** 알려 주고 각 워커가 스스로
     *     읽습니다. 읽기까지 8대로 분산됩니다.
     *
     *   원격 청킹이 맞는 경우는 시나리오 B 같은 때입니다. 읽기는 가벼운데
     *   건당 처리가 무거운 경우.
     *
     *   주의점: 마스터와 워커가 **같은 JobRepository(같은 DB)를 봐야**
     *   합니다. 그래야 워커의 StepExecution 상태를 마스터가 취합할 수
     *   있습니다. 이 지점에서 메타데이터 DB 가 단일 장애점이 됩니다.
     *
     *
     * ─────────────────────────────────────────────────────────────────
     * 종합 의사결정 순서 (13-11 의 결론)
     *
     *   1. 먼저 단일 스레드로 튜닝하십시오.
     *      인덱스, 청크 크기, rewriteBatchedStatements, fetchSize.
     *      대부분의 "느린 배치"는 병렬화 이전에 여기서 해결됩니다.
     *      62.4초를 30초로 줄이는 인덱스 하나가 8스레드보다 쌉니다.
     *
     *   2. 그래도 부족하면 로컬 파티셔닝.
     *      멀티스레드 Step 보다 빠르고(5.6배 vs 3.3배) 재시작까지 됩니다.
     *      멀티스레드 Step 은 시나리오 B 같은 I/O 대기형에만 고려하십시오.
     *
     *   3. 한 대로 안 되면 원격 파티셔닝.
     *      운영 복잡도가 급격히 올라가므로 정말 필요할 때만.
     * ─────────────────────────────────────────────────────────────────
     */
}