#Step 08 — ItemWriter

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

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

public class SettlementLogWriter implements ItemWriter<Settlement> {

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

    @Override
    public void write(Chunk<? extends Settlement> chunk) {
        log.debug("chunk size={} first={} last={}",
                chunk.size(),
                chunk.getItems().get(0).orderId(),
                chunk.getItems().get(chunk.size() - 1).orderId());
    }
}

ItemWriter<Settlement> writer = chunk -> chunk.forEach(System.out::println);

// ❌ record 에는 통하지 않습니다
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)
                """)
        .beanMapped()          // ← 여기
        .build();

@Bean
public JdbcBatchItemWriter<Settlement> settlementJdbcWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Settlement>()
            .dataSource(dataSource)
            .sql("""
                    INSERT INTO settlement
                      (order_id, customer_id, settle_date, gross_amount,
                       fee_rate, fee_amount, net_amount)
                    VALUES
                      (:orderId, :customerId, :settleDate, :grossAmount,
                       :feeRate, :feeAmount, :netAmount)
                    """)
            .itemSqlParameterSourceProvider(s -> new MapSqlParameterSource()
                    .addValue("orderId",     s.orderId())
                    .addValue("customerId",  s.customerId())
                    .addValue("settleDate",  s.settleDate())
                    .addValue("grossAmount", s.grossAmount())
                    .addValue("feeRate",     s.feeRate())
                    .addValue("feeAmount",   s.feeAmount())
                    .addValue("netAmount",   s.netAmount()))
            .assertUpdates(true)
            .build();
}

// JdbcBatchItemWriter 내부 (5.1.1)
if (assertUpdates) {
    for (int i = 0; i < updateCounts.length; i++) {
        if (updateCounts[i] == 0) {
            throw new EmptyResultDataAccessException(
                "Item " + i + " of " + updateCounts.length
                + " did not update any rows: [" + chunk.getItems().get(i) + "]", 1);
        }
    }
}

.sql("UPDATE settlement SET net_amount = :netAmount WHERE order_id = :orderId")
.assertUpdates(true)

@Bean
public FlatFileItemWriter<Settlement> settlementFileWriter() {
    DelimitedLineAggregator<Settlement> aggregator = new DelimitedLineAggregator<>();
    aggregator.setDelimiter(",");
    aggregator.setFieldExtractor(s -> new Object[]{
            s.orderId(), s.customerId(), s.settleDate(),
            s.grossAmount(), s.feeRate(), s.feeAmount(), s.netAmount()
    });

    return new FlatFileItemWriterBuilder<Settlement>()
            .name("settlementFileWriter")
            .resource(new FileSystemResource("out/settlement.csv"))
            .encoding("UTF-8")
            .lineSeparator("\n")
            .lineAggregator(aggregator)
            .headerCallback(w -> w.write(
                    "order_id,customer_id,settle_date,gross_amount,fee_rate,fee_amount,net_amount"))
            .footerCallback(w -> w.write("# generated by settlementJob"))
            .shouldDeleteIfExists(true)
            .build();
}

@Bean
public ItemWriter<Settlement> dualWriter(JdbcBatchItemWriter<Settlement> settlementJdbcWriter,
                                         FlatFileItemWriter<Settlement> settlementFileWriter) {
    CompositeItemWriter<Settlement> writer = new CompositeItemWriter<>();
    writer.setDelegates(List.of(settlementJdbcWriter, settlementFileWriter));
    return writer;
}

@Bean
public ItemWriter<Settlement> routingWriter(JdbcBatchItemWriter<Settlement> settlementJdbcWriter,
                                            FlatFileItemWriter<Settlement> vipFileWriter) {
    ClassifierCompositeItemWriter<Settlement> writer = new ClassifierCompositeItemWriter<>();
    BigDecimal vipRate = new BigDecimal("0.0150");
    writer.setClassifier((Classifier<Settlement, ItemWriter<? super Settlement>>)
            s -> vipRate.compareTo(s.feeRate()) == 0 ? vipFileWriter : settlementJdbcWriter);
    return writer;
}

public interface ItemStream {
    void open(ExecutionContext executionContext) throws ItemStreamException;
    void update(ExecutionContext executionContext) throws ItemStreamException;
    void close() throws ItemStreamException;
}

// SimpleStepBuilder.registerAsStreams() (5.1.1)
protected void registerStepListenerAsItemListener() { ... }

// build() 과정에서
if (reader instanceof ItemStream) stream((ItemStream) reader);
if (processor instanceof ItemStream) stream((ItemStream) processor);
if (writer instanceof ItemStream) stream((ItemStream) writer);

.writer(routingWriter)     // ← StepBuilder 가 받은 것은 이것뿐

public class ClassifierCompositeItemWriter<T> implements ItemWriter<T> {   // ItemStream 없음
    private Classifier<T, ItemWriter<? super T>> classifier;
    ...
}

// FlatFileItemWriter.write() 내부
if (!getOutputState().isInitialized()) {
    throw new WriterNotOpenException("Writer must be open before it can be written to");
}

@Bean
public Step routingStep(JobRepository jobRepository, PlatformTransactionManager tx,
                        ItemReader<Order> allOrdersReader,
                        ItemProcessor<Order, Settlement> settlementPipeline,
                        ItemWriter<Settlement> routingWriter,
                        FlatFileItemWriter<Settlement> vipFileWriter,
                        JdbcBatchItemWriter<Settlement> settlementJdbcWriter) {
    return new StepBuilder("routingStep", jobRepository)
            .<Order, Settlement>chunk(1000, tx)
            .reader(allOrdersReader)
            .processor(settlementPipeline)
            .writer(routingWriter)
            .stream(vipFileWriter)          // ← 숨어 있는 ItemStream 을 손으로 등록
            .build();
}

public class SettlementApiWriter implements ItemStreamWriter<Settlement> {

    private final RestClient restClient;
    private long sent = 0;

    public SettlementApiWriter(RestClient restClient) {
        this.restClient = restClient;
    }

    @Override
    public void open(ExecutionContext ctx) {
        // (1) 재시작이면 이전 위치를 복원한다
        this.sent = ctx.getLong("settlementApiWriter.sent", 0L);
    }

    @Override
    public void write(Chunk<? extends Settlement> chunk) {
        // (2) 아이템 하나씩이 아니라 청크 통째로 한 번에 보낸다
        restClient.post().uri("/settlements/bulk")
                .body(chunk.getItems())
                .retrieve().toBodilessEntity();
        sent += chunk.size();
    }

    @Override
    public void update(ExecutionContext ctx) {
        // (3) 청크 커밋마다 위치를 남긴다
        ctx.putLong("settlementApiWriter.sent", sent);
    }

    @Override
    public void close() { }
}

package com.example.batch.step08;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.support.ClassifierCompositeItemWriter;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.classify.Classifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.client.RestClient;

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

/**
 * Step 08 — ItemWriter 실습 전체 코드.
 *
 * 실행 전 준비
 *   mkdir -p ./out
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 *
 * 실행
 *   ./gradlew bootRun --args='--spring.batch.job.name=settlementJob'   # DB 만
 *   ./gradlew bootRun --args='--spring.batch.job.name=fileJob'         # DB + 파일
 *   ./gradlew bootRun --args='--spring.batch.job.name=routingJob'      # 분기
 *
 * ─────────────────────────────────────────────────────────────────────
 * [8-3] 8배 실측을 재현하려면 application.yml 의 URL 을 아래 두 개로 번갈아 두세요.
 *
 *  (A) 옵션 없음 — 70,000건 쓰기에 약 61.4초, Com_insert = 70000
 *   url: jdbc:mysql://127.0.0.1:3308/batchdb?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Seoul
 *
 *  (B) 옵션 있음 — 약 7.7초, Com_insert = 70
 *   url: jdbc:mysql://127.0.0.1:3308/batchdb?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Seoul&rewriteBatchedStatements=true
 *
 * 측정 전마다 TRUNCATE TABLE settlement; 를 하세요.
 * 안 하면 uk_settlement_order 제약 때문에 DuplicateKeyException 이 먼저 납니다.
 * ─────────────────────────────────────────────────────────────────────
 */
@Configuration
public class Practice {

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

    // ======================================================================
    // [8-1] 5.x 시그니처 — void write(Chunk<? extends T>)
    // ======================================================================

    /**
     * 4.x 는 write(List<? extends T> items) 였습니다. 5.0 에서 Chunk 로 바뀌었고
     * 이것은 소스 호환성이 깨지는 변경입니다. 4.x 코드를 붙여 넣으면
     *   error: ... does not override abstract method write(Chunk<? extends Settlement>)
     * 로 컴파일이 실패합니다. 컴파일러가 잡아 주는 "좋은 실패" 입니다.
     */
    public static class SettlementLogWriter implements ItemWriter<Settlement> {

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            log.debug("chunk size={} first={} last={}",
                    chunk.size(),
                    chunk.getItems().get(0).orderId(),
                    chunk.getItems().get(chunk.size() - 1).orderId());
        }
    }

    /** 4.x 코드를 옮길 때의 최소 어댑터. 본문은 그대로 두고 getItems() 만 끼웁니다. */
    public static class MigratedWriter implements ItemWriter<Settlement> {

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            doWrite(chunk.getItems());
        }

        /** 4.x 시절 그대로의 본문. */
        private void doWrite(List<? extends Settlement> items) {
            log.debug("{} items", items.size());
        }
    }

    /** ItemWriter 는 5.x 에서 @FunctionalInterface 입니다. 람다로도 됩니다. */
    static ItemWriter<Settlement> lambdaWriter() {
        return chunk -> chunk.forEach(s -> log.trace("{}", s.orderId()));
    }

    // ======================================================================
    // [8-2] JdbcBatchItemWriter — record 와 beanMapped()
    // ======================================================================

    /**
     * ❌ record 에는 통하지 않습니다.
     *
     * beanMapped() 는 BeanPropertyItemSqlParameterSourceProvider 를 쓰고,
     * 이것은 자바빈 규약(getOrderId())을 기대합니다. record 의 접근자는 orderId() 입니다.
     *
     * 실행 결과:
     *   org.springframework.dao.InvalidDataAccessApiUsageException:
     *     No value supplied for the SQL parameter 'orderId': Invalid property 'orderId'
     *     of bean class [com.example.batch.domain.Settlement]
     *
     * ⚠️ 이 예외가 나는 것은 settlement 의 컬럼들이 NOT NULL 이기 때문입니다.
     *    nullable 테이블이었다면 예외 없이 전 컬럼 NULL 인 70,000행이 들어가고
     *    writeCount=70000, STATUS=COMPLETED 로 모든 지표가 정상으로 보입니다.
     *    결과 테이블을 NOT NULL 로 만드는 것은 "조용한 실패를 시끄럽게 만드는 장치" 입니다.
     *
     * @Bean 이 아니므로 그냥 빌드해도 안전합니다.
     * 재현하려면 settlementJdbcWriter 자리에 손으로 바꿔 끼우세요.
     */
    public JdbcBatchItemWriter<Settlement> brokenBeanMappedWriter(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Settlement>()
                .dataSource(dataSource)
                .sql(INSERT_SQL)
                .beanMapped()
                .build();
    }

    static final String INSERT_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)
            """;

    /**
     * ✅ record 의 정답. 장황하지만 컴파일 타임에 검증됩니다.
     * record 의 컴포넌트 이름을 바꾸면 여기서 컴파일 에러가 납니다.
     * beanMapped() 는 리팩터링을 따라오지 못합니다.
     */
    @Bean
    public JdbcBatchItemWriter<Settlement> settlementJdbcWriter(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Settlement>()
                .dataSource(dataSource)
                .sql(INSERT_SQL)
                .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()))
                // [8-4] 기본값 true. UPDATE writer 에서 "0건 갱신"을 잡아 줍니다.
                .assertUpdates(true)
                .build();
    }

    // ======================================================================
    // [8-4] assertUpdates — 무엇을 잡고 무엇을 못 잡는가
    // ======================================================================

    /**
     * UPDATE writer. WHERE 에 걸리는 행이 없으면 갱신 건수가 0 이고,
     * assertUpdates(true) 가 EmptyResultDataAccessException 을 던집니다.
     *
     *   Item 137 of 1000 did not update any rows: [Settlement[orderId=999999, ...]]
     *
     * ⚠️ 구멍 두 개
     *  (1) rewriteBatchedStatements=true 로 UPDATE 배치가 다중 문장으로 재작성되면
     *      드라이버가 Statement.SUCCESS_NO_INFO(-2) 를 돌려줍니다.
     *      JdbcBatchItemWriter 의 검사는 `updateCounts[i] == 0` 뿐이라 -2 는 통과합니다.
     *      성능 옵션을 켜는 순간 안전장치가 조용히 꺼집니다.
     *  (2) "1건 기대했는데 70,000건 갱신됨"은 잡지 못합니다.
     *      WHERE 절을 빠뜨린 UPDATE 는 갱신 건수가 0 이 아니기 때문입니다.
     *      옵션으로 못 막습니다. UPDATE 의 WHERE 에 유니크 키가 있는지 리뷰하세요.
     */
    public JdbcBatchItemWriter<Settlement> settlementUpdateWriter(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Settlement>()
                .dataSource(dataSource)
                .sql("UPDATE settlement SET net_amount = :netAmount WHERE order_id = :orderId")
                .itemSqlParameterSourceProvider(s -> new MapSqlParameterSource()
                        .addValue("orderId", s.orderId())
                        .addValue("netAmount", s.netAmount()))
                .assertUpdates(true)
                .build();
    }

    // ======================================================================
    // [8-5] FlatFileItemWriter — CSV
    // ======================================================================

    /**
     * BeanWrapperFieldExtractor 는 beanMapped() 와 같은 이유로 record 에 통하지 않습니다.
     * FieldExtractor 를 람다로 씁니다.
     *
     * headerCallback 은 open() 에서, footerCallback 은 close() 에서 실행됩니다([8-7]).
     * 이 사실이 [8-8] 함정의 뿌리입니다.
     */
    @Bean
    public FlatFileItemWriter<Settlement> settlementFileWriter() {
        return csvWriter("settlementFileWriter", "out/settlement.csv",
                "# generated by settlementJob");
    }

    @Bean
    public FlatFileItemWriter<Settlement> vipFileWriter() {
        return csvWriter("vipFileWriter", "out/vip-settlement.csv",
                "# vip settlements: generated by fileJob");
    }

    private static FlatFileItemWriter<Settlement> csvWriter(String name, String path, String footer) {
        DelimitedLineAggregator<Settlement> aggregator = new DelimitedLineAggregator<>();
        aggregator.setDelimiter(",");
        aggregator.setFieldExtractor(s -> new Object[]{
                s.orderId(), s.customerId(), s.settleDate(),
                s.grossAmount(), s.feeRate(), s.feeAmount(), s.netAmount()
        });

        return new FlatFileItemWriterBuilder<Settlement>()
                .name(name)                       // ExecutionContext 키의 접두사가 됩니다
                .resource(new FileSystemResource(path))
                .encoding("UTF-8")                // 플랫폼 기본값에 맡기지 마세요
                .lineSeparator("\n")              // 윈도우에서 \r 이 섞이는 사고 방지
                .lineAggregator(aggregator)
                .headerCallback(w -> w.write(
                        "order_id,customer_id,settle_date,gross_amount,"
                                + "fee_rate,fee_amount,net_amount"))
                .footerCallback(w -> w.write(footer))
                .shouldDeleteIfExists(true)
                .transactional(true)              // 기본값. 청크 커밋 시점에 flush
                .build();
    }

    // ======================================================================
    // [8-6] CompositeItemWriter — 같은 청크를 여러 곳에
    // ======================================================================

    /**
     * 모든 델리게이트가 같은 청크를 받습니다.
     * writeCount 는 70000 한 번만 올라갑니다. 두 곳에 썼다고 140000 이 되지 않습니다.
     *
     * ✅ CompositeItemWriter 는 ItemStreamWriter 를 구현하므로
     *    open/update/close 를 모든 델리게이트에 전파합니다. .stream() 이 필요 없습니다.
     */
    @Bean
    public ItemWriter<Settlement> dualWriter(JdbcBatchItemWriter<Settlement> settlementJdbcWriter,
                                             FlatFileItemWriter<Settlement> settlementFileWriter) {
        CompositeItemWriter<Settlement> writer = new CompositeItemWriter<>();
        writer.setDelegates(List.of(settlementJdbcWriter, settlementFileWriter));
        return writer;
    }

    // ======================================================================
    // [8-6][8-8] ClassifierCompositeItemWriter — 아이템별 분기
    // ======================================================================

    /**
     * VIP(fee_rate 0.0150) 15,000건은 파일로, 나머지 55,000건은 DB 로.
     *
     * ❌ ClassifierCompositeItemWriter 는 ItemStream 을 구현하지 않습니다.
     *    안에 숨은 vipFileWriter 의 open/update/close 가 한 번도 불리지 않습니다.
     *    -> 파일 0줄, 또는 WriterNotOpenException.
     *    -> routingStep 의 .stream(vipFileWriter) 로 손수 등록해야 합니다.
     *
     * 이름이 비슷한 CompositeItemWriter 는 전파하는데 이쪽은 안 합니다.
     * 이 비대칭이 함정의 본질입니다.
     */
    @Bean
    public ItemWriter<Settlement> routingWriter(JdbcBatchItemWriter<Settlement> settlementJdbcWriter,
                                                FlatFileItemWriter<Settlement> vipFileWriter) {
        ClassifierCompositeItemWriter<Settlement> writer = new ClassifierCompositeItemWriter<>();
        BigDecimal vipRate = new BigDecimal("0.0150");

        Classifier<Settlement, ItemWriter<? super Settlement>> classifier =
                s -> vipRate.compareTo(s.feeRate()) == 0 ? vipFileWriter : settlementJdbcWriter;

        writer.setClassifier(classifier);
        return writer;
    }

    // ======================================================================
    // [8-9] 커스텀 ItemStreamWriter
    // ======================================================================

    /**
     * 외부 API 로 쓰는 writer. 실제 엔드포인트가 없으므로 그대로 실행하면
     * 커넥션 예외가 납니다. 구조를 읽는 용도입니다.
     *
     * 규칙 셋
     *  (1) write() 안에서 아이템 단위 루프로 원격 호출 금지 — [8-3] 의 8배 문제와 동형입니다.
     *  (2) open() 에서 복원하고 update() 에서 저장 — 안 하면 재시작 시 중복 전송됩니다.
     *  (3) 멱등하게 — 청크가 롤백되면 같은 청크가 다시 옵니다.
     *      REST 호출은 롤백되지 않습니다. 이미 보낸 것은 보낸 것입니다.
     *      order_id 같은 멱등 키를 실어 보내고 수신 측이 중복을 무시하게 하세요.
     */
    public static class SettlementApiWriter implements ItemStreamWriter<Settlement> {

        private static final String CTX_KEY = "settlementApiWriter.sent";

        private final RestClient restClient;
        private long sent = 0;

        public SettlementApiWriter(RestClient restClient) {
            this.restClient = restClient;
        }

        @Override
        public void open(ExecutionContext ctx) {
            this.sent = ctx.getLong(CTX_KEY, 0L);      // (2) 재시작 복원
        }

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            restClient.post().uri("/settlements/bulk")  // (1) 청크 통째로 한 번에
                    .body(chunk.getItems())
                    .retrieve().toBodilessEntity();
            sent += chunk.size();
        }

        @Override
        public void update(ExecutionContext ctx) {
            ctx.putLong(CTX_KEY, sent);                // (2) 청크 커밋마다 저장
        }

        @Override
        public void close() {
        }
    }

    // ======================================================================
    // Reader / Processor / Listener
    // ======================================================================

    @Bean
    public ItemReader<Order> allOrdersReader(DataSource dataSource) {
        return new JdbcCursorItemReaderBuilder<Order>()
                .name("allOrdersReader")
                .dataSource(dataSource)
                .sql("SELECT order_id, customer_id, amount, status, ordered_at "
                        + "FROM orders ORDER BY order_id")
                .rowMapper((rs, rowNum) -> new Order(
                        rs.getLong("order_id"),
                        rs.getInt("customer_id"),
                        rs.getBigDecimal("amount"),
                        rs.getString("status"),
                        rs.getTimestamp("ordered_at").toLocalDateTime()))
                .fetchSize(1000)
                .build();
    }

    /** Step 07 의 파이프라인을 그대로 씁니다. 10만 건 읽고 3만 건 필터 -> 7만 건 출력. */
    @Bean
    public ItemProcessor<Order, Settlement> settlementPipeline(
            org.springframework.jdbc.core.JdbcTemplate jdbcTemplate) {
        var step1 = new com.example.batch.step07.Practice.CompletedOnlyProcessor();
        var step2 = new com.example.batch.step07.Practice.GradeFeeSettlementProcessor(jdbcTemplate);
        return order -> {
            Order filtered = step1.process(order);
            return filtered == null ? null : step2.process(filtered);
        };
    }

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

    // ======================================================================
    // Step / Job
    // ======================================================================

    /** [8-3] 8배 측정용. DB 에만 씁니다. */
    @Bean
    public Step settlementStep(JobRepository jobRepository, PlatformTransactionManager tx,
                               ItemReader<Order> allOrdersReader,
                               ItemProcessor<Order, Settlement> settlementPipeline,
                               JdbcBatchItemWriter<Settlement> settlementJdbcWriter,
                               StepExecutionListener countLogger) {
        return new StepBuilder("settlementStep", jobRepository)
                .<Order, Settlement>chunk(1000, tx)
                .reader(allOrdersReader)
                .processor(settlementPipeline)
                .writer(settlementJdbcWriter)
                .listener(countLogger)
                .build();
    }

    /** [8-6] DB + 파일. CompositeItemWriter 라 .stream() 이 필요 없습니다. */
    @Bean
    public Step dualStep(JobRepository jobRepository, PlatformTransactionManager tx,
                         ItemReader<Order> allOrdersReader,
                         ItemProcessor<Order, Settlement> settlementPipeline,
                         ItemWriter<Settlement> dualWriter,
                         StepExecutionListener countLogger) {
        return new StepBuilder("dualStep", jobRepository)
                .<Order, Settlement>chunk(1000, tx)
                .reader(allOrdersReader)
                .processor(settlementPipeline)
                .writer(dualWriter)
                .listener(countLogger)
                .build();
    }

    /**
     * [8-8] 함정 재현 스텝.
     *
     * 순서를 지키세요.
     *  1. .stream(vipFileWriter) 가 주석인 채로 실행 -> 파일 0줄 / WriterNotOpenException
     *     그리고 BATCH_STEP_EXECUTION_CONTEXT 에 vipFileWriter.written 키가 없는 것 확인
     *  2. 주석을 풀고 다시 실행 -> 15,002줄, 컨텍스트에 vipFileWriter.written=15000
     *
     * 고쳐진 코드만 보면 이 함정은 배워지지 않습니다.
     */
    @Bean
    public Step routingStep(JobRepository jobRepository, PlatformTransactionManager tx,
                            ItemReader<Order> allOrdersReader,
                            ItemProcessor<Order, Settlement> settlementPipeline,
                            ItemWriter<Settlement> routingWriter,
                            FlatFileItemWriter<Settlement> vipFileWriter,
                            StepExecutionListener countLogger) {
        return new StepBuilder("routingStep", jobRepository)
                .<Order, Settlement>chunk(1000, tx)
                .reader(allOrdersReader)
                .processor(settlementPipeline)
                .writer(routingWriter)
                // .stream(vipFileWriter)      // ← 1단계에서는 주석, 2단계에서 해제
                .listener(countLogger)
                .build();
    }

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

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

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

package com.example.batch.step08;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.core.Step;
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.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.support.ClassifierCompositeItemWriter;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.List;

/**
 * Step 08 — 연습문제 7문항.
 *
 * 준비
 *   mkdir -p ./out
 *   mysql -h127.0.0.1 -P3308 -uroot -proot1234 batchdb -e "TRUNCATE TABLE settlement;"
 *
 * 규칙
 *  - "여기에 작성:" 자리를 채우세요.
 *  - 문제 2·3·4·6 은 반드시 "예측 먼저, 실행 나중" 입니다.
 *
 * 정답은 Solution.java.
 */
public class Exercise {

    // ======================================================================
    // 문제 1 — 4.x writer 를 5.x 시그니처로 마이그레이션
    // ======================================================================
    /**
     * 아래는 Spring Batch 4.x 스타일 writer 입니다. 그대로 두면 컴파일되지 않습니다.
     * doWrite(List) 본문은 건드리지 말고, 시그니처만 5.x 로 바꾸세요.
     *
     * 컴파일 에러 메시지를 먼저 예측해 보세요.
     *   여기에 작성(예상 에러):
     */
    public static class Q1_LegacyWriter implements ItemWriter<Settlement> {

        // 여기에 작성: 5.x 시그니처로 write() 를 구현하세요.

        /** 4.x 시절 본문. 수정하지 마세요. */
        private void doWrite(List<? extends Settlement> items) {
            System.out.println("wrote " + items.size() + " items");
        }
    }

    // ======================================================================
    // 문제 2 — record 에 맞는 JdbcBatchItemWriter
    // ======================================================================
    /**
     * (a) 아래 writer 에서 .beanMapped() 를 쓰면 어떤 예외가 납니까?
     *     예외 클래스 : ______________________
     *     메시지 요지 : ______________________
     *     그 이유     : 여기에 작성:
     *
     * (b) settlement 테이블의 컬럼이 전부 nullable 이었다면 무슨 일이 벌어집니까?
     *     writeCount = ______, Job STATUS = ______, 데이터 = ______
     *     여기에 작성(설명):
     *
     * (c) record 에 맞는 writer 를 완성하세요.
     */
    public static class Q2_RecordWriter {

        public JdbcBatchItemWriter<Settlement> build(DataSource dataSource) {
            return new JdbcBatchItemWriterBuilder<Settlement>()
                    .dataSource(dataSource)
                    .sql("""
                            INSERT INTO settlement
                              (order_id, customer_id, settle_date, gross_amount,
                               fee_rate, fee_amount, net_amount)
                            VALUES
                              (:orderId, :customerId, :settleDate, :grossAmount,
                               :feeRate, :feeAmount, :netAmount)
                            """)
                    // 여기에 작성: itemSqlParameterSourceProvider 를 람다로 채우세요.

                    .assertUpdates(true)
                    .build();
        }
    }

    // ======================================================================
    // 문제 3 — rewriteBatchedStatements 실측 (측정 절차 문제)
    // ======================================================================
    /**
     * 코드가 아니라 절차를 수행하는 문제입니다. 두 번 측정하세요.
     *
     * 각 측정마다:
     *   1) application.yml 의 datasource.url 을 (A) 또는 (B) 로 바꾼다
     *      (A) ...serverTimezone=Asia/Seoul
     *      (B) ...serverTimezone=Asia/Seoul&rewriteBatchedStatements=true
     *   2) mysql ... -e "TRUNCATE TABLE settlement;"
     *   3) mysql ... -uroot -proot1234 -e "FLUSH STATUS;"
     *   4) ./gradlew bootRun --args='--spring.batch.job.name=settlementJob'
     *   5) mysql ... -uroot -proot1234 -e "SHOW GLOBAL STATUS LIKE 'Com_insert';"
     *
     * 결과를 채우세요.
     *   ┌───────────────┬──────────────┬─────────────┬────────────┐
     *   │               │ Step 실행시간 │ Com_insert  │ writeCount │
     *   ├───────────────┼──────────────┼─────────────┼────────────┤
     *   │ (A) 옵션 없음  │              │             │            │
     *   │ (B) 옵션 있음  │              │             │            │
     *   └───────────────┴──────────────┴─────────────┴────────────┘
     *   배수: 약 ______ 배
     *
     * (a) 두 실행의 writeCount / Job STATUS / 결과 행 수가 같습니까? 답:
     *     여기에 작성:
     * (b) 그렇다면 이 문제를 "모니터링만으로" 발견할 수 있습니까? 답과 대안:
     *     여기에 작성:
     */

    // ======================================================================
    // 문제 4 — assertUpdates 가 못 잡는 두 상황
    // ======================================================================
    /**
     * ⚠️ 이 문제는 settlement 데이터를 파괴합니다. 실행 후 TRUNCATE 하고 재생성하세요.
     *
     * (a) 아래 writer 는 WHERE 절이 없습니다.
     *     70,000건 청크를 흘리면 최종적으로 settlement 는 어떤 상태가 됩니까?
     *     여기에 작성(예측):
     *     assertUpdates(true) 가 이것을 잡습니까? 답과 이유:
     *     여기에 작성:
     *
     * (b) rewriteBatchedStatements=true 인 상태에서 UPDATE 배치가 다중 문장으로
     *     재작성되면 드라이버는 각 항목에 어떤 값을 돌려줍니까?
     *     값 = ______ (상수 이름과 숫자)
     *     JdbcBatchItemWriter 의 검사는 `updateCounts[i] == 0` 입니다.
     *     이 값이 그 검사를 통과합니까? 답과 그 의미:
     *     여기에 작성:
     *
     * (c) 두 상황 각각의 대안 방어책을 적으세요.
     *     (a) 의 방어책: 여기에 작성:
     *     (b) 의 방어책: 여기에 작성:
     */
    public static class Q4_OverUpdate {

        public JdbcBatchItemWriter<Settlement> dangerous(DataSource dataSource) {
            return new JdbcBatchItemWriterBuilder<Settlement>()
                    .dataSource(dataSource)
                    .sql("UPDATE settlement SET fee_rate = :feeRate")   // WHERE 가 없다
                    .itemSqlParameterSourceProvider(s -> new MapSqlParameterSource()
                            .addValue("feeRate", s.feeRate()))
                    .assertUpdates(true)
                    .build();
        }
    }

    // ======================================================================
    // 문제 5 — FlatFileItemWriter 와 header / footer
    // ======================================================================
    /**
     * out/exercise-settlement.csv 를 만드세요.
     *  - 구분자 ","
     *  - 헤더: order_id,customer_id,settle_date,gross_amount,fee_rate,fee_amount,net_amount
     *  - 푸터: # total lines above = 70000
     *  - 인코딩 UTF-8, 줄 구분자 "\n", 기존 파일은 삭제 후 새로 쓰기
     *
     * 실행 후 `wc -l out/exercise-settlement.csv` 의 결과를 예측하세요.
     *   예측 = ______줄
     *   그 이유: 여기에 작성:
     *
     * 힌트: FieldExtractor 로 BeanWrapperFieldExtractor 를 쓰면 안 되는 이유는
     *       문제 2 의 beanMapped() 와 완전히 같습니다.
     */
    public static class Q5_FileWriter {

        public FlatFileItemWriter<Settlement> build() {
            DelimitedLineAggregator<Settlement> aggregator = new DelimitedLineAggregator<>();
            aggregator.setDelimiter(",");
            // 여기에 작성: setFieldExtractor

            return new FlatFileItemWriterBuilder<Settlement>()
                    .name("exerciseFileWriter")
                    .resource(new FileSystemResource("out/exercise-settlement.csv"))
                    // 여기에 작성: encoding / lineSeparator / lineAggregator /
                    //             headerCallback / footerCallback / shouldDeleteIfExists

                    .build();
        }
    }

    // ======================================================================
    // 문제 6 — ClassifierCompositeItemWriter 와 .stream()
    // ======================================================================
    /**
     * VIP(fee_rate 0.0150) 15,000건은 파일로, 나머지 55,000건은 DB 로 보냅니다.
     *
     * 아래 Step 에는 .stream() 이 빠져 있습니다.
     *
     * 1단계: 이 상태로 그대로 실행하세요. (실패가 정상입니다)
     *   Job STATUS       = ______
     *   DB 행 수         = ______
     *   파일 줄 수       = ______
     *   예외가 났다면 클래스 = ______________________
     *   BATCH_STEP_EXECUTION_CONTEXT 에 vipFileWriter.written 키가 있습니까? ______
     *
     * 2단계: 왜 이렇게 되는지 적으세요.
     *   여기에 작성:
     *   (힌트: ClassifierCompositeItemWriter 는 무엇을 구현하지 "않는가"?
     *          CompositeItemWriter 와 비교하세요.)
     *
     * 3단계: 고치세요.
     */
    public static class Q6_RoutingStep {

        public Step build(JobRepository jobRepository, PlatformTransactionManager tx,
                          ItemReader<Order> reader,
                          ItemProcessor<Order, Settlement> processor,
                          ClassifierCompositeItemWriter<Settlement> routingWriter,
                          FlatFileItemWriter<Settlement> vipFileWriter) {
            return new StepBuilder("q6RoutingStep", jobRepository)
                    .<Order, Settlement>chunk(1000, tx)
                    .reader(reader)
                    .processor(processor)
                    .writer(routingWriter)
                    // 여기에 작성(3단계):

                    .build();
        }
    }

    // ======================================================================
    // 문제 7 — 커스텀 ItemStreamWriter 의 생명주기 (서술형)
    // ======================================================================
    /**
     * 아래 writer 를 완성하고, 다음 질문에 답하세요.
     *
     * (a) open() 을 비워 두면 무엇이 깨집니까?
     *     여기에 작성:
     * (b) update() 를 비워 두면 무엇이 깨집니까?
     *     여기에 작성:
     * (c) (a) 와 (b) 중 어느 쪽이 더 발견하기 어렵습니까? 왜입니까?
     *     여기에 작성:
     * (d) 청크가 롤백되면 이미 전송한 아이템은 어떻게 됩니까? 방어책은?
     *     여기에 작성:
     */
    public static class Q7_CustomWriter implements ItemStreamWriter<Settlement> {

        private static final String CTX_KEY = "q7Writer.sent";
        private long sent = 0;

        @Override
        public void open(ExecutionContext ctx) {
            // 여기에 작성:
        }

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            // 실제 전송 대신 카운트만 합니다.
            sent += chunk.size();
        }

        @Override
        public void update(ExecutionContext ctx) {
            // 여기에 작성:
        }

        @Override
        public void close() {
        }
    }
}

package com.example.batch.step08;

import com.example.batch.domain.Order;
import com.example.batch.domain.Settlement;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.support.ClassifierCompositeItemWriter;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.List;

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

    // ======================================================================
    // 정답 1 — 5.x 시그니처로 마이그레이션
    // ======================================================================
    /*
     * 예상 컴파일 에러:
     *   error: Q1_LegacyWriter is not abstract and does not override abstract method
     *          write(Chunk<? extends Settlement>) in ItemWriter
     *   error: method does not override or implement a method from a supertype
     *
     * 4.x 의 write(List<? extends T>) 는 5.0 에서 write(Chunk<? extends T>) 로 바뀌었습니다.
     * 소스 호환성이 깨지는 변경이고, 이것은 "좋은 실패" 입니다 — 컴파일러가 잡아 주니까요.
     * 이 코스에서 다루는 대부분의 함정은 컴파일러가 못 잡는 것들입니다.
     *
     * 마이그레이션은 어댑터 한 줄이면 끝납니다. 본문(doWrite)은 손대지 않습니다.
     * Chunk 로 바뀐 실질적 이득은 getSkips() / getErrors() 처럼
     * "아이템 말고 다른 정보"를 함께 실을 수 있다는 점입니다. List 로는 불가능했습니다.
     */
    public static class A1_MigratedWriter implements ItemWriter<Settlement> {

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            doWrite(chunk.getItems());
        }

        private void doWrite(List<? extends Settlement> items) {
            System.out.println("wrote " + items.size() + " items");
        }
    }

    // ======================================================================
    // 정답 2 — record 와 beanMapped()
    // ======================================================================
    /*
     * (a)
     *   예외 클래스: org.springframework.dao.InvalidDataAccessApiUsageException
     *   메시지 요지: No value supplied for the SQL parameter 'orderId':
     *               Invalid property 'orderId' of bean class
     *               [com.example.batch.domain.Settlement]
     *
     *   이유: beanMapped() 는 BeanPropertyItemSqlParameterSourceProvider 를 쓰고,
     *         이 클래스는 BeanWrapper 를 통해 "자바빈 규약"으로 값을 읽습니다.
     *         자바빈 규약의 읽기 접근자는 getOrderId() 입니다.
     *         record 의 접근자는 orderId() — get 접두사가 없습니다.
     *         그래서 BeanWrapper 는 orderId 라는 읽을 수 있는 프로퍼티가 없다고 판단합니다.
     *
     * (b) settlement 가 nullable 이었다면 — 이쪽이 진짜 무서운 경우입니다.
     *   writeCount = 70000
     *   Job STATUS = COMPLETED
     *   데이터     = 70,000행이 들어가 있고 order_id 부터 net_amount 까지 전부 NULL
     *
     *   로그도 정상, 카운터도 정상, 행 수도 정상입니다.
     *   "정산이 안 맞는다"는 문의가 며칠 뒤에 옵니다.
     *
     *   실제로 재현해 보려면:
     *     CREATE TABLE settlement_nullable LIKE settlement;
     *     ALTER TABLE settlement_nullable
     *       MODIFY gross_amount DECIMAL(12,2) NULL,
     *       MODIFY net_amount   DECIMAL(12,2) NULL,
     *       DROP INDEX uk_settlement_order;
     *   그리고 이 테이블로 beanMapped() writer 를 돌려 보세요.
     *
     *   교훈: 결과 테이블의 컬럼은 되도록 NOT NULL 로 만드세요.
     *         제약은 성능 장치이기 이전에 "조용한 실패를 시끄럽게 만드는 장치" 입니다.
     *         project 스펙의 settlement 가 전 컬럼 NOT NULL + uk_settlement_order 인 것은
     *         우연이 아닙니다.
     *
     * (c) 정답 코드는 아래.
     *     장황해 보이지만 record 의 컴포넌트 이름을 바꾸면 여기서 컴파일 에러가 납니다.
     *     beanMapped() 는 리팩터링을 따라오지 못하고, 런타임에야 알려 줍니다.
     *
     *     읽기 쪽도 대칭입니다: BeanPropertyRowMapper 는 record 를 못 채우고,
     *     DataClassRowMapper 가 생성자 기반으로 채웁니다(Step 06).
     */
    public static class A2_RecordWriter {

        public JdbcBatchItemWriter<Settlement> build(DataSource dataSource) {
            return new JdbcBatchItemWriterBuilder<Settlement>()
                    .dataSource(dataSource)
                    .sql("""
                            INSERT INTO settlement
                              (order_id, customer_id, settle_date, gross_amount,
                               fee_rate, fee_amount, net_amount)
                            VALUES
                              (:orderId, :customerId, :settleDate, :grossAmount,
                               :feeRate, :feeAmount, :netAmount)
                            """)
                    .itemSqlParameterSourceProvider(s -> new MapSqlParameterSource()
                            .addValue("orderId",     s.orderId())
                            .addValue("customerId",  s.customerId())
                            .addValue("settleDate",  s.settleDate())
                            .addValue("grossAmount", s.grossAmount())
                            .addValue("feeRate",     s.feeRate())
                            .addValue("feeAmount",   s.feeAmount())
                            .addValue("netAmount",   s.netAmount()))
                    .assertUpdates(true)
                    .build();
        }
    }

    // ======================================================================
    // 정답 3 — rewriteBatchedStatements 실측
    // ======================================================================
    /*
     *   ┌───────────────┬──────────────┬─────────────┬────────────┐
     *   │               │ Step 실행시간 │ Com_insert  │ writeCount │
     *   ├───────────────┼──────────────┼─────────────┼────────────┤
     *   │ (A) 옵션 없음  │  1m 1s 412ms │      70000  │      70000 │
     *   │ (B) 옵션 있음  │     7s 702ms │         70  │      70000 │
     *   └───────────────┴──────────────┴─────────────┴────────────┘
     *   61.4초 -> 7.7초. 약 8배.
     *
     * 왜 8배인가 — 분해해 봅니다.
     *   줄어든 왕복 = 70,000 - 70 = 69,930회
     *   줄어든 시간 = 61.4 - 7.7 = 53.7초
     *   왕복 1회당  = 53.7s / 69,930 ≈ 0.77ms
     *   로컬 도커 MySQL 에 대한 TCP 왕복 + 파싱 비용으로 타당한 값입니다.
     *
     *   즉 사라진 것은 DB 가 한 일이 아니라 "왕복" 입니다.
     *   InnoDB 가 실제로 삽입한 행 수는 양쪽 다 70,000 으로 같습니다.
     *   그래서 서버를 키워도, 인덱스를 지워도 8배가 안 나옵니다.
     *   병목이 DB 가 아니기 때문입니다.
     *
     * (a) writeCount, Job STATUS, 결과 행 수가 모두 같습니까?
     *     네, 완전히 같습니다. 70000 / COMPLETED / 70000.
     *
     * (b) 모니터링만으로 발견 가능한가?
     *     아니오. 어떤 에러도 경고도 남지 않습니다.
     *     그래서 "우리 배치는 원래 한 시간 걸려요" 가 굳어집니다.
     *
     *     대안 진단법: 서버 쪽 Com_insert 를 재세요.
     *       FLUSH STATUS;  -> Job 실행 -> SHOW GLOBAL STATUS LIKE 'Com_insert';
     *     이 값이 "처리 건수와 같으면" 배치 쓰기가 배치가 아닌 것입니다.
     *     정상이라면 처리건수 / 청크크기 정도의 값이 나와야 합니다(70,000/1,000 = 70).
     *
     *     참고: PostgreSQL·Oracle 드라이버에는 이 옵션이 없습니다.
     *           기본적으로 배치를 제대로 묶어 보냅니다. MySQL 만의 기본값 문제입니다.
     */

    // ======================================================================
    // 정답 4 — assertUpdates 의 두 구멍
    // ======================================================================
    /*
     * (a) WHERE 절 없는 UPDATE
     *   각 아이템마다 settlement 70,000행이 전부 갱신됩니다.
     *   청크의 1,000개 아이템이 순서대로 전체를 덮으므로,
     *   최종적으로 남는 값은 "마지막에 처리된 아이템의 fee_rate" 하나입니다.
     *   70,000행 전체가 같은 fee_rate 를 갖게 됩니다.
     *
     *   assertUpdates(true) 가 잡습니까? 아니오.
     *   검사는 `updateCounts[i] == 0` 뿐입니다. 여기서는 각 항목이 70000 입니다.
     *   0 이 아니므로 통과합니다. writeCount 도 70000 으로 정상입니다.
     *
     *   이것이 assertUpdates 의 본질적 한계입니다.
     *   "너무 적게 갱신됨"만 잡고 "너무 많이 갱신됨"은 개념적으로 잡을 수 없습니다.
     *
     * (b) SUCCESS_NO_INFO
     *   값 = java.sql.Statement.SUCCESS_NO_INFO, 숫자로는 -2.
     *
     *   rewriteBatchedStatements=true 일 때 UPDATE/DELETE 배치는 멀티 VALUES 로
     *   합칠 수 없어서 여러 문장을 이어 보내는 형태로 재작성됩니다.
     *   이때 드라이버는 개별 문장의 갱신 건수를 분리해 낼 수 없어 -2 를 채웁니다.
     *
     *   `updateCounts[i] == 0` 검사를 통과합니까? 네, 통과합니다. -2 != 0 이니까요.
     *
     *   의미: 성능 옵션을 켜는 순간 UPDATE writer 의 안전장치가 조용히 꺼집니다.
     *   8-3 에서 "가장 가성비 좋은 한 줄"이라고 한 옵션이,
     *   8-4 의 안전장치를 무력화합니다. 두 절을 따로 읽으면 이 상호작용을 놓칩니다.
     *
     * (c) 대안 방어책
     *   (a) 의 방어책 — 코드 리뷰 규칙으로 못 박습니다.
     *       "ItemWriter 의 UPDATE/DELETE 문에는 반드시 유니크 키가 WHERE 에 있어야 한다."
     *       settlement 라면 order_id (uk_settlement_order) 입니다.
     *       자동화하려면 SQL 문자열을 정적 검사하는 테스트를 붙이세요.
     *
     *   (b) 의 방어책 — 갱신 건수를 믿지 말고 Step 뒤에서 결과를 직접 검증합니다.
     *       아래 A4_VerifyTasklet 처럼 Tasklet Step 을 붙이고,
     *       기대 건수와 다르면 예외를 던져 Job 을 FAILED 로 만듭니다.
     *       Step 10 의 흐름 제어를 쓰면 검증 실패 시 보정 Step 으로 분기시킬 수도 있습니다.
     */
    public static class A4_VerifyTasklet implements Tasklet {

        private final JdbcTemplate jdbcTemplate;
        private final long expected;

        public A4_VerifyTasklet(JdbcTemplate jdbcTemplate, long expected) {
            this.jdbcTemplate = jdbcTemplate;
            this.expected = expected;
        }

        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext ctx) {
            Long actual = jdbcTemplate.queryForObject(
                    "SELECT COUNT(*) FROM settlement", Long.class);
            if (actual == null || actual != expected) {
                // 조용히 틀리느니 시끄럽게 실패하는 편이 낫습니다.
                throw new IllegalStateException(
                        "정산 건수 불일치: expected=" + expected + ", actual=" + actual);
            }
            return RepeatStatus.FINISHED;
        }
    }

    // ======================================================================
    // 정답 5 — FlatFileItemWriter
    // ======================================================================
    /*
     * wc -l 예측 = 70002 줄.
     *   데이터 70,000행 + headerCallback 1줄 + footerCallback 1줄.
     *
     * FieldExtractor 를 람다로 쓰는 이유는 문제 2 와 완전히 같습니다.
     * BeanWrapperFieldExtractor 는 이름 그대로 BeanWrapper 를 쓰므로
     * record 의 orderId() 를 읽지 못합니다.
     *
     * 옵션 중 실무에서 실제로 사고를 내는 것 둘:
     *  - encoding 을 지정하지 않으면 플랫폼 기본 인코딩이 쓰입니다.
     *    개발자 맥에서는 UTF-8, 운영 리눅스 컨테이너에서 POSIX 로케일이면 US-ASCII 가
     *    되어 한글이 '?' 로 깨집니다. 로컬에서는 절대 재현되지 않습니다.
     *  - lineSeparator 를 지정하지 않으면 System.lineSeparator() 입니다.
     *    윈도우에서 만든 파일에 \r\n 이 섞여 수신 시스템의 파서가 마지막 필드에
     *    \r 을 붙여 읽습니다. 이것도 조용한 실패입니다.
     * 둘 다 "명시하면 끝나는" 문제라 명시하는 것이 정답입니다.
     */
    public static class A5_FileWriter {

        public FlatFileItemWriter<Settlement> build() {
            DelimitedLineAggregator<Settlement> aggregator = new DelimitedLineAggregator<>();
            aggregator.setDelimiter(",");
            aggregator.setFieldExtractor(s -> new Object[]{
                    s.orderId(), s.customerId(), s.settleDate(),
                    s.grossAmount(), s.feeRate(), s.feeAmount(), s.netAmount()
            });

            return new FlatFileItemWriterBuilder<Settlement>()
                    .name("exerciseFileWriter")
                    .resource(new FileSystemResource("out/exercise-settlement.csv"))
                    .encoding("UTF-8")
                    .lineSeparator("\n")
                    .lineAggregator(aggregator)
                    .headerCallback(w -> w.write(
                            "order_id,customer_id,settle_date,gross_amount,"
                                    + "fee_rate,fee_amount,net_amount"))
                    .footerCallback(w -> w.write("# total lines above = 70000"))
                    .shouldDeleteIfExists(true)
                    .build();
        }
    }

    // ======================================================================
    // 정답 6 — .stream() 수동 등록
    // ======================================================================
    /*
     * 1단계 관찰 결과
     *   Job STATUS   = FAILED (WriterNotOpenException) 또는
     *                  COMPLETED (파일은 0줄) — vipFileWriter 가 @Bean 이라
     *                  afterPropertiesSet() 이 불려 빈 파일만 만들어진 경우
     *   DB 행 수     = 55000  (JdbcBatchItemWriter 는 ItemStream 이 아니라 멀쩡합니다)
     *   파일 줄 수   = 0
     *   예외 클래스  = org.springframework.batch.item.WriterNotOpenException
     *                  "Writer must be open before it can be written to"
     *   vipFileWriter.written 키 = 없음
     *
     * 2단계 — 왜 이렇게 되는가
     *   StepBuilder 는 .reader()/.processor()/.writer() 로 "직접 넘긴" 객체가
     *   ItemStream 이면 자동으로 stream() 에 등록합니다.
     *   여기서 직접 넘긴 것은 routingWriter 하나뿐이고,
     *   ClassifierCompositeItemWriter 는 ItemStream 을 구현하지 않습니다.
     *
     *     public class ClassifierCompositeItemWriter<T> implements ItemWriter<T>
     *                                                              ^^^^^^^^^^^^ 이것뿐
     *
     *   그래서 안에 숨은 vipFileWriter 의 콜백이 하나도 불리지 않습니다.
     *   빠진 것은 셋이고 각각 다른 피해를 냅니다.
     *
     *     open()   미호출 -> 파일이 열리지 않음 + headerCallback 미실행
     *     update() 미호출 -> ExecutionContext 에 written 위치가 안 남음
     *                        => 재시작하면 파일을 처음부터 다시 씀.
     *                           리더는 중단 지점부터 이어가는데 파일은 처음부터라
     *                           파일에는 뒤쪽 데이터만 남습니다. DB 는 맞고 파일은 틀립니다.
     *     close()  미호출 -> footerCallback 미실행 + 마지막 버퍼 유실
     *                        (transactional=true 라 출력이 버퍼에 있습니다)
     *
     *   진짜 원인은 "비대칭" 입니다.
     *     CompositeItemWriter           : ItemStreamWriter 구현 -> 델리게이트에 전파함
     *     ClassifierCompositeItemWriter : ItemWriter 만 구현    -> 전파 안 함
     *   이름이 한 단어 다를 뿐인데 정반대로 동작합니다.
     *   "CompositeItemWriter 로 잘 되던 코드"를 분기가 필요해져 바꾸는 순간
     *   파일이 조용히 비기 시작합니다. 코드 리뷰에서는 클래스 이름만 바뀐 것으로 보입니다.
     *
     * 3단계 — 고치기: .stream(vipFileWriter) 한 줄.
     *   JdbcBatchItemWriter 는 ItemStream 이 아니므로 등록할 필요가 없습니다
     *   (파일 핸들 같은 상태가 없습니다).
     *
     *   고친 뒤 검증:
     *     DB 55000 + 파일 15002줄 (= 15000 + 헤더 + 푸터) = 70000
     *     SELECT SHORT_CONTEXT FROM BATCH_STEP_EXECUTION_CONTEXT
     *       ORDER BY STEP_EXECUTION_ID DESC LIMIT 1;
     *     -> {"vipFileWriter.written":15000, "allOrdersReader.read.count":100000}
     *
     *   습관으로 만들 점검 3줄:
     *     1. ItemStream 구현체가 .reader()/.writer() 에 직접 들어갔는가?
     *     2. 아니라면 .stream(...) 으로 손수 등록했는가?
     *     3. 한 번 돌리고 컨텍스트에 <name>.written 키가 있는가?
     *   3번은 10초면 됩니다. 안 하면 재시작이 필요해지는 그날까지 아무도 모릅니다.
     */
    public static class A6_RoutingStep {

        public Step build(JobRepository jobRepository, PlatformTransactionManager tx,
                          ItemReader<Order> reader,
                          ItemProcessor<Order, Settlement> processor,
                          ClassifierCompositeItemWriter<Settlement> routingWriter,
                          FlatFileItemWriter<Settlement> vipFileWriter) {
            return new StepBuilder("q6RoutingStep", jobRepository)
                    .<Order, Settlement>chunk(1000, tx)
                    .reader(reader)
                    .processor(processor)
                    .writer(routingWriter)
                    .stream(vipFileWriter)      // ← 숨어 있는 ItemStream 을 수동 등록
                    .build();
        }
    }

    // ======================================================================
    // 정답 7 — 커스텀 ItemStreamWriter 의 생명주기
    // ======================================================================
    /*
     * (a) open() 을 비워 두면
     *     재시작 시 이전에 전송한 건수를 복원하지 못해 sent 가 0 부터 시작합니다.
     *     이 예제처럼 sent 를 "몇 건 보냈나" 리포트에만 쓴다면 리포트만 틀립니다.
     *     그러나 sent 를 "여기부터 보내면 된다"는 위치로 쓴다면 중복 전송이 일어납니다.
     *     외부 리소스를 여는 writer(파일·소켓·커넥션)라면 아예 열리지 않아
     *     첫 write() 에서 즉시 터집니다.
     *
     * (b) update() 를 비워 두면
     *     ExecutionContext 에 위치가 저장되지 않습니다.
     *     open() 이 복원할 것이 애초에 없으므로, (a) 와 같은 결과가 되지만
     *     발현 시점이 다릅니다.
     *
     * (c) 어느 쪽이 더 발견하기 어려운가 — update() 쪽입니다.
     *
     *     ┌───────────┬────────────────────┬──────────────────────────┐
     *     │           │ 첫 실행            │ 재시작                   │
     *     ├───────────┼────────────────────┼──────────────────────────┤
     *     │ open 누락 │ 즉시 실패하기도 함 │ 위치 복원 안 됨          │
     *     │ update 누락│ 완벽하게 성공     │ 처음부터 다시 전송(중복) │
     *     └───────────┴────────────────────┴──────────────────────────┘
     *
     *     update() 누락은 첫 실행이 완벽하게 성공합니다.
     *     단위 테스트도 통과하고, 스테이징에서도 통과하고, 운영 첫날도 통과합니다.
     *     재시작이 필요해지는 날 — 즉 이미 장애 대응 중인 순간에 — 두 번째 장애로
     *     모습을 드러냅니다. 최악의 타이밍입니다.
     *
     *     이 코스가 반복해 말하는 종류의 버그입니다:
     *     문법 에러는 금방 고치지만, 에러 없이 조용히 틀리는 코드가 진짜 위험합니다.
     *
     * (d) 청크 롤백 시
     *     DB writer 는 함께 롤백됩니다. 외부 API 호출은 롤백되지 않습니다.
     *     청크의 900번째에서 예외가 나면 DB 는 깨끗이 되돌아가지만
     *     API 쪽에는 이미 1,000건이 들어가 있습니다. 재시도하면 2,000건이 됩니다.
     *
     *     방어책은 멱등 키입니다.
     *     order_id 처럼 아이템마다 고유한 값을 요청에 실어 보내고,
     *     수신 측이 같은 키를 두 번 받으면 무시하게 만듭니다.
     *     이 코스의 settlement.order_id 에 걸린 UNIQUE 제약이 바로 그 DB 버전입니다.
     *     "한 주문에 정산 한 번"을 스키마가 보장하므로, 배치를 두 번 돌려도
     *     정산이 두 배가 되는 대신 DuplicateKeyException 으로 시끄럽게 실패합니다.
     */
    public static class A7_CustomWriter implements ItemStreamWriter<Settlement> {

        private static final String CTX_KEY = "q7Writer.sent";
        private long sent = 0;

        @Override
        public void open(ExecutionContext ctx) {
            this.sent = ctx.getLong(CTX_KEY, 0L);   // 재시작 복원
        }

        @Override
        public void write(Chunk<? extends Settlement> chunk) {
            // 실제 구현이라면 여기서 chunk.getItems() 를 통째로 한 번에 보냅니다.
            // 아이템 단위 루프로 원격 호출하면 8-3 의 8배 문제가 그대로 재현됩니다.
            sent += chunk.size();
        }

        @Override
        public void update(ExecutionContext ctx) {
            ctx.putLong(CTX_KEY, sent);             // 청크 커밋마다 위치 저장
        }

        @Override
        public void close() {
        }
    }
}