#Step 14 — 운영: 스케줄링 · 모니터링 · 최종 프로젝트

@Component
@RequiredArgsConstructor
public class SettlementScheduler {

    private final JobLauncher jobLauncher;
    private final Job dailySettlementJob;

    @Scheduled(cron = "0 0 2 * * *", zone = "Asia/Seoul")     // 매일 새벽 2시
    public void runDailySettlement() throws Exception {
        LocalDate targetDate = LocalDate.now().minusDays(1);   // 어제치

        JobParameters params = new JobParametersBuilder()
                .addString("date", targetDate.toString())      // identifying
                .addLong("launchedAt", System.currentTimeMillis(), false)  // non-identifying
                .toJobParameters();

        jobLauncher.run(dailySettlementJob, params);
    }
}

@Configuration
public class QuartzConfig {

    @Bean
    public JobDetail settlementJobDetail() {
        return JobBuilder.newJob(SettlementQuartzJob.class)
                .withIdentity("settlementQuartzJob")
                .storeDurably()
                .build();
    }

    @Bean
    public Trigger settlementTrigger(JobDetail settlementJobDetail) {
        return TriggerBuilder.newTrigger()
                .forJob(settlementJobDetail)
                .withIdentity("settlementTrigger")
                .withSchedule(CronScheduleBuilder
                        .cronSchedule("0 0 2 * * ?")
                        .inTimeZone(TimeZone.getTimeZone("Asia/Seoul"))
                        // ⚠️ 놓친 실행을 어떻게 할 것인가 — 아래 함정 참조
                        .withMisfireHandlingInstructionDoNothing())
                .build();
    }
}

@Component
@RequiredArgsConstructor
public class SettlementQuartzJob extends QuartzJobBean {

    private final JobLauncher jobLauncher;
    private final Job dailySettlementJob;

    @Override
    protected void executeInternal(JobExecutionContext context) {
        // ...JobParameters 를 만들어 jobLauncher.run(...)
    }
}

@Component
public class BatchExitCodeGenerator implements ExitCodeGenerator {

    private final JobExplorer jobExplorer;

    @Override
    public int getExitCode() {
        // 0 = 성공, 1 = 실패, 2 = 데이터 없음(정산 대상 0건) 등
        // 운영 파이프라인이 재시도 여부를 판단하는 근거가 됩니다.
        return ...;
    }
}

public boolean isAlreadyRunning(String jobName) {
    Set<JobExecution> running = jobExplorer.findRunningJobExecutions(jobName);
    return !running.isEmpty();
}

@Scheduled(cron = "0 0 2 * * *", zone = "Asia/Seoul")
public void runDailySettlement() throws Exception {
    if (isAlreadyRunning("dailySettlementJob")) {
        log.warn("이전 정산 배치가 아직 실행 중입니다. 이번 실행을 건너뜁니다.");
        return;
    }
    // ...
}

public boolean tryAcquireLock(String jobName, LocalDate date) {
    try {
        jdbcTemplate.update("""
                INSERT INTO batch_job_lock (job_name, target_date, acquired_at, holder)
                VALUES (?, ?, NOW(), ?)
                """, jobName, date, InetAddress.getLocalHost().getHostName());
        return true;
    } catch (DuplicateKeyException e) {
        return false;       // 다른 프로세스가 이미 잡았습니다
    }
}

@Bean
public StepExecutionListener metricsListener(MeterRegistry registry) {
    return new StepExecutionListener() {
        @Override
        public ExitStatus afterStep(StepExecution se) {
            registry.gauge("settlement.written.amount",
                    Tags.of("date", se.getJobParameters().getString("date")),
                    fetchTotalNetAmount(se));
            return se.getExitStatus();      // Step 12 의 규칙
        }
    };
}

@Bean
@StepScope
public JdbcPagingItemReader<Order> dailyOrderReader(
        DataSource dataSource,
        @Value("#{jobParameters['date']}") String date) {

    MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
    provider.setSelectClause("o.order_id, o.customer_id, o.amount, o.status, o.ordered_at");
    provider.setFromClause("FROM orders o LEFT JOIN settlement s ON s.order_id = o.order_id");
    provider.setWhereClause("""
            WHERE o.status = 'COMPLETED'
              AND DATE(o.ordered_at) = :date
              AND s.order_id IS NULL
            """);
    // 유니크한 정렬 키. Step 06 의 함정을 피합니다.
    provider.setSortKeys(Map.of("o.order_id", Order.ASCENDING));

    return new JdbcPagingItemReaderBuilder<Order>()
            .name("dailyOrderReader")               // ExecutionContext 키 접두사
            .dataSource(dataSource)
            .queryProvider(provider)
            .parameterValues(Map.of("date", date))
            .pageSize(500)
            .rowMapper(new DataClassRowMapper<>(Order.class))
            .build();
}

public class SettlementDecider implements JobExecutionDecider {
    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        long written = stepExecution == null ? 0 : stepExecution.getWriteCount();
        return new FlowExecutionStatus(written > 0 ? "HAS_DATA" : "NOTHING");
    }
}

@Bean
public Job dailySettlementJob(JobRepository jobRepository,
                              Step dailySettlementStep,
                              Step reportStep,
                              SettlementDecider decider) {
    return new JobBuilder("dailySettlementJob", jobRepository)
            .listener(new SettlementJobListener())
            .start(dailySettlementStep)
            .next(decider)
                .on("HAS_DATA").to(reportStep)
            .from(decider)
                .on("NOTHING").end()
            .end()
            .build();
}

package com.example.batch.step14;

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.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.listener.JobExecutionListener;
import org.springframework.batch.core.listener.SkipListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.io.BufferedWriter;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.InetAddress;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Step 14 — 운영: 스케줄링 · 모니터링 · 최종 프로젝트
 *
 * 본문 14-1 ~ 14-8 의 모든 예제 + 종합 실습 Job 전체.
 * 이 코스에서 가장 긴 실습 파일입니다.
 *
 * ─────────────────────────────────────────────────────────────────────────
 * ⚠️ 이 스텝은 bootRun 이 아니라 bootJar + java -jar 로 실습하십시오.
 *
 *   종료 코드 확인이 실습의 절반인데, bootRun 은 Gradle 이 종료 코드를
 *   가려 버립니다.
 *
 *     ./gradlew clean bootJar
 *     java -jar build/libs/spring-batch5-lab-1.0.0.jar \
 *       --spring.batch.job.enabled=true \
 *       --spring.batch.job.name=dailySettlementJob \
 *       date=2025-03-01
 *     echo $?
 *
 * ─────────────────────────────────────────────────────────────────────────
 * ⚠️ [14-1] 먼저 application.yml 을 바꾸십시오.
 *
 *   spring:
 *     batch:
 *       job:
 *         enabled: false        # ← 이걸 안 바꾸면 이 파일의 Job 들이
 *                               #   부팅 때 우르르 돌아 실습이 뒤엉킵니다
 *     task:
 *       scheduling:
 *         pool:
 *           size: 5             # @Scheduled 기본 풀 크기 1 → 5
 *
 *   management:
 *     endpoints:
 *       web:
 *         exposure:
 *           include: health, metrics, prometheus
 * ─────────────────────────────────────────────────────────────────────────
 */
public class Practice {

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

    // =====================================================================
    // [14-2] @Scheduled — 가장 단순한 스케줄
    //
    // ⚠️ cron 이 새벽 2시라 실습 중에는 절대 돌지 않습니다.
    //    테스트하려면 runNow() 를 직접 호출하십시오.
    //
    //    cron 을 "0/10 * * * * *" 로 바꿔 두고 잊으면
    //    10초마다 정산이 돕니다. 실습 후 반드시 되돌리십시오.
    // =====================================================================

    // @Component
    public static class SettlementScheduler {

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

        private final JobLauncher jobLauncher;
        private final Job dailySettlementJob;
        private final JobExplorer jobExplorer;

        public SettlementScheduler(JobLauncher jobLauncher,
                                   Job dailySettlementJob,
                                   JobExplorer jobExplorer) {
            this.jobLauncher = jobLauncher;
            this.dailySettlementJob = dailySettlementJob;
            this.jobExplorer = jobExplorer;
        }

        @Scheduled(cron = "0 0 2 * * *", zone = "Asia/Seoul")
        public void runDailySettlement() throws Exception {
            runFor(LocalDate.now().minusDays(1));       // 어제치
        }

        /** 실습용 수동 실행. 스케줄을 기다리지 않고 바로 돌립니다. */
        public void runNow(LocalDate targetDate) throws Exception {
            runFor(targetDate);
        }

        private void runFor(LocalDate targetDate) throws Exception {
            // 2차 방어 — 이전 실행이 아직 도는 중이면 건너뜁니다.
            Set<JobExecution> running =
                    jobExplorer.findRunningJobExecutions("dailySettlementJob");
            if (!running.isEmpty()) {
                log.warn("이전 정산 배치가 아직 실행 중입니다. 이번 실행을 건너뜁니다. running={}",
                        running);
                return;
            }

            log.info("일일 정산 시작: date={}", targetDate);

            JobParameters params = new JobParametersBuilder()
                    // identifying — JOB_KEY 에 포함됩니다.
                    .addString("date", targetDate.toString())
                    // non-identifying — JOB_KEY 에 영향 없음.
                    // RunIdIncrementer 를 쓰면 중복 차단이 무력화되므로,
                    // 감사 추적용 값은 이렇게 false 로 넣습니다.
                    .addLong("launchedAt", System.currentTimeMillis(), false)
                    .toJobParameters();

            jobLauncher.run(dailySettlementJob, params);
        }
    }

    // =====================================================================
    // [14-5] 중복 실행 방지 — 3단 방어
    //
    // ★ 이 절이 이 스텝에서 가장 중요합니다.
    //    정산 배치가 두 번 돌면 그건 곧 돈입니다.
    // =====================================================================

    /**
     * 3차 방어 — DB 유니크 락.
     *
     * 1차(JobInstance 중복 차단)와 2차(RUNNING 확인)는 "확인 후 실행"
     * 구조라 경쟁 상태를 막지 못합니다. 확인과 실행 사이에 틈이 있습니다.
     *
     * DB 의 PK 제약에는 그 틈이 없습니다. 두 프로세스가 정확히 동시에
     * INSERT 해도 DB 가 원자적으로 하나만 통과시킵니다.
     */
    // @Component
    public static class JobLockService {

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

        private final JdbcTemplate jdbcTemplate;

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

        public static final String DDL = """
                CREATE TABLE IF NOT EXISTS batch_job_lock (
                  job_name    VARCHAR(100) NOT NULL,
                  target_date DATE         NOT NULL,
                  acquired_at DATETIME     NOT NULL,
                  holder      VARCHAR(100) NOT NULL,
                  PRIMARY KEY (job_name, target_date)   -- ← 방어의 실체
                ) ENGINE=InnoDB;
                """;

        public boolean tryAcquire(String jobName, LocalDate date) {
            try {
                jdbcTemplate.update("""
                        INSERT INTO batch_job_lock
                          (job_name, target_date, acquired_at, holder)
                        VALUES (?, ?, NOW(), ?)
                        """, jobName, date, hostname());
                log.info("락 획득 성공: {}/{}", jobName, date);
                return true;
            } catch (DuplicateKeyException e) {
                log.warn("락 획득 실패 — 다른 인스턴스가 실행 중입니다. 종료합니다.");
                return false;
            }
        }

        public void release(String jobName, LocalDate date) {
            jdbcTemplate.update(
                    "DELETE FROM batch_job_lock WHERE job_name = ? AND target_date = ?",
                    jobName, date);
        }

        private String hostname() {
            try {
                return InetAddress.getLocalHost().getHostName();
            } catch (Exception e) {
                return "unknown";
            }
        }
    }

    /**
     * 좀비 실행 정리기.
     *
     * ⚠️ 배치 서버가 kill -9 로 죽으면 BATCH_JOB_EXECUTION 에
     *    STATUS='STARTED', END_TIME=NULL 인 행이 영원히 남습니다.
     *    아무도 정리해 주지 않습니다.
     *
     *    그러면 2차 방어(findRunningJobExecutions)가 "아직 실행 중"으로
     *    판단해 **정산이 영원히 건너뛰어집니다.**
     *    중복을 막으려던 장치가 실행 자체를 막는 역설입니다.
     *
     * ⚠️ 실습 중에는 @PostConstruct 를 주석 처리해 두었습니다.
     *    연습문제 2 에서 좀비를 일부러 만들어야 하는데, 자동 정리가
     *    켜져 있으면 재현이 안 되기 때문입니다.
     */
    // @Component
    public static class ZombieCleaner {

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

        /**
         * 좀비 판정 기준 시간.
         *
         * ⚠️ 이 값의 판단이 가장 까다롭습니다.
         *    배치의 최대 실행 시간보다 넉넉히 잡아야 합니다.
         *    정상 실행 중인 배치를 좀비로 오인해 FAILED 로 만들면
         *    그게 훨씬 큰 사고입니다.
         */
        private static final int ZOMBIE_THRESHOLD_HOURS = 6;

        private final JdbcTemplate jdbcTemplate;

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

        // @PostConstruct      // ← 연습문제 2 를 풀 때는 주석 처리한 채로 두십시오
        public void cleanUpOnStartup() {
            List<Long> zombies = jdbcTemplate.queryForList("""
                    SELECT JOB_EXECUTION_ID FROM BATCH_JOB_EXECUTION
                    WHERE STATUS IN ('STARTED','STARTING')
                      AND END_TIME IS NULL
                      AND START_TIME < NOW() - INTERVAL ? HOUR
                    """, Long.class, ZOMBIE_THRESHOLD_HOURS);

            for (Long id : zombies) {
                jdbcTemplate.update("""
                        UPDATE BATCH_JOB_EXECUTION
                           SET STATUS = 'FAILED', EXIT_CODE = 'FAILED',
                               END_TIME = NOW(), LAST_UPDATED = NOW(),
                               EXIT_MESSAGE = '좀비 실행 자동 정리 (기동 시 감지)'
                         WHERE JOB_EXECUTION_ID = ?
                        """, id);
                log.warn("좀비 실행을 정리했습니다: JOB_EXECUTION_ID={}", id);
            }
        }
    }

    // =====================================================================
    // [14-8] 종합 실습 — 일일 주문 정산 배치
    //
    // Step 01~13 의 모든 요소를 하나로 조립합니다.
    // 각 빈에 어느 스텝에서 온 요소인지 표시해 두었으니,
    // 코스를 복습하는 지도로 쓰십시오.
    //
    //   흐름:
    //     ① dailySettlementStep  (청크 500)
    //           ↓ ExitStatus
    //     ② SettlementDecider
    //        ├ NOTHING  → end()
    //        └ HAS_DATA → ③ reportStep (CSV 출력)
    // =====================================================================

    // @Configuration
    public static class DailySettlementJobConfig {

        // ── Job 조립 ─────────────────────────────────────────────────
        @Bean
        public Job dailySettlementJob(JobRepository jobRepository,
                                      Step dailySettlementStep,
                                      Step reportStep,
                                      SettlementDecider settlementDecider) {
            return new JobBuilder("dailySettlementJob", jobRepository)
                    .listener(new SettlementJobListener())          // Step 12
                    .start(dailySettlementStep)
                    .next(settlementDecider)                        // Step 10
                        .on("HAS_DATA").to(reportStep)
                    .from(settlementDecider)
                        .on("NOTHING").end()
                    .end()
                    .build();
        }

        // ── ① 정산 Step ──────────────────────────────────────────────
        @Bean
        public Step dailySettlementStep(JobRepository jobRepository,
                                        PlatformTransactionManager txManager,
                                        JdbcPagingItemReader<Order> dailyOrderReader,
                                        DataSource dataSource) {
            return new StepBuilder("dailySettlementStep", jobRepository)
                    .<Order, Settlement>chunk(500, txManager)       // Step 05
                    .reader(dailyOrderReader)                       // Step 06
                    .processor(new GradeFeeProcessor())             // Step 07
                    .writer(idempotentSettlementWriter(dataSource)) // Step 08
                    .faultTolerant()                                // Step 11
                    .skip(IllegalArgumentException.class)
                    .skipLimit(50)
                    .listener(new BadOrderSkipListener(dataSource)) // Step 12
                    .build();
        }

        /**
         * Reader — 이미 정산된 주문을 안티 조인으로 제외합니다.
         *
         * Step 11 연습문제 3 의 결론입니다. 쓰기에서 UNIQUE 충돌로
         * 터뜨리는 대신 애초에 읽지 않으면, 스캔 모드에 빠지지 않아
         * 재실행이 훨씬 빠릅니다. 그리고 이것이 중복 정산의
         * 최종 방어선 역할도 합니다.
         *
         * ⚠️ @StepScope 가 필수입니다 (Step 09).
         *    없으면 기동 시점에 jobParameters 가 없어 SpEL 이 터집니다.
         * ⚠️ 리턴 타입이 인터페이스가 아니라 구현체입니다 (Step 09).
         *    ItemReader<Order> 로 선언하면 스코프 프록시가 ItemStream 을
         *    잃어 재시작이 조용히 파손됩니다.
         */
        @Bean
        @StepScope
        public JdbcPagingItemReader<Order> dailyOrderReader(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {

            MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
            provider.setSelectClause(
                    "o.order_id, o.customer_id, o.amount, o.status, o.ordered_at");
            provider.setFromClause(
                    "FROM orders o LEFT JOIN settlement s ON s.order_id = o.order_id");
            provider.setWhereClause("""
                    WHERE o.status = 'COMPLETED'
                      AND DATE(o.ordered_at) = :date
                      AND s.order_id IS NULL
                    """);
            // ⚠️ 유니크한 정렬 키 (Step 06 의 함정).
            //    유니크하지 않으면 페이지 경계에서 데이터가 조용히 사라집니다.
            provider.setSortKeys(Map.of(
                    "o.order_id", org.springframework.batch.item.database.Order.ASCENDING));

            return new JdbcPagingItemReaderBuilder<Order>()
                    .name("dailyOrderReader")       // ExecutionContext 키 접두사 (Step 11)
                    .dataSource(dataSource)
                    .queryProvider(provider)
                    .parameterValues(Map.of("date", date))
                    .pageSize(500)
                    .rowMapper(new DataClassRowMapper<>(Order.class))
                    .build();
        }

        /**
         * Writer — 멱등 INSERT (Step 08).
         *
         * record 는 자바빈이 아니므로 beanMapped() 를 못 씁니다.
         * 람다로 직접 매핑합니다.
         */
        @Bean
        public JdbcBatchItemWriter<Settlement> idempotentSettlementWriter(
                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)
                         """)
                    .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();
        }

        // ── ② 분기 Decider ───────────────────────────────────────────
        @Bean
        public SettlementDecider settlementDecider() {
            return new SettlementDecider();
        }

        // ── ③ 리포트 Step ────────────────────────────────────────────
        @Bean
        @StepScope
        public ReportTasklet reportTasklet(
                DataSource dataSource,
                @Value("#{jobParameters['date']}") String date) {
            return new ReportTasklet(dataSource, date);
        }

        @Bean
        public Step reportStep(JobRepository jobRepository,
                               PlatformTransactionManager txManager,
                               ReportTasklet reportTasklet) {
            return new StepBuilder("reportStep", jobRepository)
                    .tasklet(reportTasklet, txManager)      // Step 04
                    .build();
        }
    }

    /**
     * 등급별 수수료 Processor (Step 07).
     *
     * ⚠️ 상태가 없습니다. Step 13 의 스레드 안전성 원칙입니다.
     */
    public static class GradeFeeProcessor implements ItemProcessor<Order, Settlement> {

        // customer_id % 4 → 등급 수수료율. 시드 규칙과 일치합니다.
        private static final BigDecimal[] FEE_RATES = {
                new BigDecimal("0.0350"),   // 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) {
            if (order.amount().signum() < 0) {
                throw new IllegalArgumentException(
                        "정산 금액이 음수입니다: order_id=" + order.order_id());
            }
            BigDecimal feeRate = FEE_RATES[order.customerId() % 4];
            BigDecimal gross = order.amount();
            // 반올림 시점이 중요합니다 (Step 07). 곱한 뒤 즉시 2자리로 고정합니다.
            BigDecimal fee = gross.multiply(feeRate).setScale(2, RoundingMode.HALF_UP);
            return new Settlement(order.order_id(), order.customerId(),
                    order.orderedAt().toLocalDate(),
                    gross, feeRate, fee, gross.subtract(fee));
        }
    }

    /** 정산 건수에 따라 분기 (Step 10). */
    public static class SettlementDecider implements JobExecutionDecider {
        @Override
        public FlowExecutionStatus decide(JobExecution jobExecution,
                                          StepExecution stepExecution) {
            long written = stepExecution == null ? 0 : stepExecution.getWriteCount();
            return new FlowExecutionStatus(written > 0 ? "HAS_DATA" : "NOTHING");
        }
    }

    /** 불량 주문 기록 (Step 12). 커밋 후에 호출되므로 롤백되지 않습니다. */
    public static class BadOrderSkipListener implements SkipListener<Order, Settlement> {

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

        private final JdbcTemplate jdbcTemplate;

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

        @Override
        public void onSkipInProcess(Order item, Throwable t) {
            log.warn("[SKIP] order_id={}, 사유={}", item.order_id(), t.getMessage());
            jdbcTemplate.update("""
                    INSERT INTO s14_bad_order (order_id, reason, occurred_at)
                    VALUES (?, ?, NOW())
                    """, item.order_id(), t.getMessage());
        }
    }

    /** Job 시작/종료 알림 (Step 12). */
    public static class SettlementJobListener implements JobExecutionListener {

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

        @Override
        public void beforeJob(JobExecution jobExecution) {
            log.info(">>> 일일 정산 시작. date={}",
                    jobExecution.getJobParameters().getString("date"));
        }

        @Override
        public void afterJob(JobExecution jobExecution) {
            long millis = Duration.between(
                    jobExecution.getStartTime(), jobExecution.getEndTime()).toMillis();
            log.info(">>> 일일 정산 종료. status={}, 소요={}ms",
                    jobExecution.getStatus(), millis);

            // ⚠️ 상태 분기 없이 "완료" 알림을 보내면 실패해도 알림이 갑니다.
            if (jobExecution.getStatus() == BatchStatus.FAILED) {
                try {
                    log.error(">>> 정산 실패! 원인={}",
                            jobExecution.getAllFailureExceptions());
                } catch (Exception e) {
                    // afterJob 의 예외는 삼켜지므로 여기서 반드시 잡습니다.
                    log.error(">>> 알림 전송 실패", e);
                }
            }
        }
    }

    /**
     * 정산 요약을 CSV 로 내보냅니다 (Step 04 Tasklet + Step 08 파일 출력).
     *
     * ⚠️ output/ 디렉터리가 없으면 FileNotFoundException 이 납니다.
     *    아래에서 createDirectories 로 처리하지만, 권한 문제는
     *    직접 확인해야 합니다. mkdir output 을 먼저 해 두십시오.
     */
    public static class ReportTasklet
            implements org.springframework.batch.core.step.tasklet.Tasklet {

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

        private final JdbcTemplate jdbcTemplate;
        private final String date;

        public ReportTasklet(DataSource dataSource, String date) {
            this.jdbcTemplate = new JdbcTemplate(dataSource);
            this.date = date;
        }

        @Override
        public RepeatStatus execute(org.springframework.batch.core.StepContribution contribution,
                                    org.springframework.batch.core.scope.context.ChunkContext ctx)
                throws Exception {

            Path dir = Path.of("output");
            Files.createDirectories(dir);
            Path file = dir.resolve("settlement-" + date + ".csv");

            List<Map<String, Object>> rows = jdbcTemplate.queryForList("""
                    SELECT order_id, customer_id, gross_amount, fee_rate,
                           fee_amount, net_amount
                    FROM settlement
                    WHERE settle_date = ?
                    ORDER BY order_id
                    """, date);

            try (BufferedWriter w = Files.newBufferedWriter(file)) {
                w.write("order_id,customer_id,gross_amount,fee_rate,fee_amount,net_amount");
                w.newLine();
                for (Map<String, Object> r : rows) {
                    w.write("%s,%s,%s,%s,%s,%s".formatted(
                            r.get("order_id"), r.get("customer_id"),
                            r.get("gross_amount"), r.get("fee_rate"),
                            r.get("fee_amount"), r.get("net_amount")));
                    w.newLine();
                }
            }

            // Tasklet 은 카운터를 자동으로 세지 않습니다 (Step 04).
            contribution.incrementWriteCount(rows.size());
            log.info("리포트 생성 완료: {} ({}줄)", file, rows.size() + 1);
            return RepeatStatus.FINISHED;
        }
    }

    // =====================================================================
    // [14-6] 운영 분석 SQL 모음
    //
    // 그대로 복사해 mysql 클라이언트에 붙이거나,
    // 모니터링 대시보드의 쿼리로 쓰십시오.
    // =====================================================================

    public static class OperationalQueries {

        public static final String RECENT_FAILURES = """
                SELECT ji.JOB_NAME, je.JOB_EXECUTION_ID AS exec_id, je.START_TIME,
                       je.STATUS, LEFT(je.EXIT_MESSAGE, 80) AS reason
                FROM BATCH_JOB_EXECUTION je
                JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
                WHERE je.STATUS = 'FAILED'
                ORDER BY je.START_TIME DESC
                LIMIT 5;
                """;

        public static final String DURATION_TREND = """
                SELECT DATE(je.START_TIME) AS d,
                       COUNT(*) AS runs,
                       ROUND(AVG(TIMESTAMPDIFF(SECOND, je.START_TIME, je.END_TIME)), 1) AS avg_sec,
                       MAX(TIMESTAMPDIFF(SECOND, je.START_TIME, je.END_TIME)) AS max_sec
                FROM BATCH_JOB_EXECUTION je
                JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
                WHERE ji.JOB_NAME = 'dailySettlementJob' AND je.END_TIME IS NOT NULL
                GROUP BY DATE(je.START_TIME)
                ORDER BY d DESC LIMIT 7;
                """;

        public static final String SLOWEST_STEPS = """
                SELECT se.STEP_NAME, COUNT(*) AS runs,
                       ROUND(AVG(TIMESTAMPDIFF(SECOND, se.START_TIME, se.END_TIME)), 1) AS avg_sec,
                       SUM(se.READ_COUNT) AS total_read,
                       SUM(se.WRITE_COUNT) AS total_write,
                       SUM(se.ROLLBACK_COUNT) AS rollbacks
                FROM BATCH_STEP_EXECUTION se
                WHERE se.END_TIME IS NOT NULL
                GROUP BY se.STEP_NAME
                ORDER BY avg_sec DESC;
                """;

        /**
         * 정합성 등식 검증 (Step 01).
         *   READ = WRITE + FILTER + SKIP
         * 빈 결과가 정상입니다.
         */
        public static final String INTEGRITY_CHECK = """
                SELECT se.STEP_EXECUTION_ID,
                       se.READ_COUNT, se.WRITE_COUNT, se.FILTER_COUNT,
                       se.READ_SKIP_COUNT + se.PROCESS_SKIP_COUNT
                         + se.WRITE_SKIP_COUNT AS skips,
                       se.READ_COUNT - se.WRITE_COUNT - se.FILTER_COUNT
                         - (se.READ_SKIP_COUNT + se.PROCESS_SKIP_COUNT
                            + se.WRITE_SKIP_COUNT) AS unexplained
                FROM BATCH_STEP_EXECUTION se
                WHERE se.STATUS = 'COMPLETED'
                HAVING unexplained <> 0;
                """;

        /**
         * ⚠️ 위 등식이 잡지 못하는 유실이 있습니다 (Step 13).
         *    프레임워크 카운터는 "프레임워크가 본 것"일 뿐입니다.
         *    반드시 업무 데이터 쪽에서도 검증하십시오. 이게 최종 방어선입니다.
         */
        public static final String BUSINESS_INTEGRITY_CHECK = """
                SELECT ? AS target_date,
                       (SELECT COUNT(*) FROM orders
                         WHERE status = 'COMPLETED' AND DATE(ordered_at) = ?) AS 대상,
                       (SELECT COUNT(*) FROM settlement
                         WHERE settle_date = ?)                               AS 정산;
                """;

        /** 좀비 실행 탐지 (14-5). */
        public static final String FIND_ZOMBIES = """
                SELECT je.JOB_EXECUTION_ID, ji.JOB_NAME, je.START_TIME, je.STATUS
                FROM BATCH_JOB_EXECUTION je
                JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
                WHERE je.STATUS IN ('STARTED','STARTING')
                  AND je.END_TIME IS NULL
                  AND je.START_TIME < NOW() - INTERVAL 6 HOUR;
                """;
    }

    // =====================================================================
    // 실습 준비 DDL
    // =====================================================================

    public static final String SETUP_DDL = """
            CREATE TABLE IF NOT EXISTS s14_bad_order (
              id          BIGINT       NOT NULL AUTO_INCREMENT PRIMARY KEY,
              order_id    BIGINT       NOT NULL,
              reason      VARCHAR(500) NOT NULL,
              occurred_at DATETIME     NOT NULL
            ) ENGINE=InnoDB;

            CREATE TABLE IF NOT EXISTS batch_job_lock (
              job_name    VARCHAR(100) NOT NULL,
              target_date DATE         NOT NULL,
              acquired_at DATETIME     NOT NULL,
              holder      VARCHAR(100) NOT NULL,
              PRIMARY KEY (job_name, target_date)
            ) ENGINE=InnoDB;
            """;
}

package com.example.batch.step14;

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.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.time.LocalDate;

/**
 * Step 14 — 연습문제 (6문제)
 *
 * 정답은 Solution.java. 먼저 직접 풀어 보십시오.
 *
 * ─────────────────────────────────────────────────────────────────────────
 * ⚠️ 이 스텝의 문제는 셸에서 하는 작업의 비중이 큽니다.
 *    bootRun 이 아니라 bootJar + java -jar 로 실습하십시오.
 *
 *      ./gradlew clean bootJar
 *
 * 사전 준비:
 *   - application.yml 의 spring.batch.job.enabled 를 false 로
 *   - Practice.SETUP_DDL 로 s14_bad_order, batch_job_lock 테이블 생성
 *   - Practice.ZombieCleaner 의 @PostConstruct 는 주석 처리된 상태 유지
 *     (문제 2 에서 좀비를 만들어야 하므로)
 * ─────────────────────────────────────────────────────────────────────────
 */
public class Exercise {

    // =====================================================================
    // 문제 1. 종료 코드 확인하기
    //
    // ★ 코드를 거의 안 쓰는 문제입니다. 그런데 배포 전 체크리스트에서
    //    가장 자주 빠지는 항목입니다.
    //
    // (a) dailySettlementJob 을 date=2025-03-01 로 실행하고
    //     종료 코드를 확인하십시오.
    //
    //       java -jar build/libs/spring-batch5-lab-1.0.0.jar \
    //         --spring.batch.job.enabled=true \
    //         --spring.batch.job.name=dailySettlementJob date=2025-03-01
    //       echo $?
    //
    //     종료 코드: ____
    //
    // (b) 같은 명령을 한 번 더 실행하십시오 (JobInstanceAlreadyComplete).
    //     종료 코드: ____
    //
    // (c) BatchLabApplication.main 에서 SpringApplication.exit(...) 감싸기를
    //     제거하고 (b) 를 다시 하십시오.
    //     종료 코드: ____
    //
    // (d) ⚠️ (c) 의 결과가 이 문제의 핵심입니다.
    //     배치가 실패했는데 종료 코드가 무엇입니까?
    //     이것이 왜 위험합니까? 로그를 봐도 발견되지 않는 이유는?
    //
    // (e) 실습이 끝나면 main 을 원래대로 되돌리십시오.
    // =====================================================================

    // (a)(b)(c) 종료 코드
    // 여기에 작성:
    //

    // (d) 왜 위험한가
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 2. 좀비 실행 만들고, 2차 방어가 무력화되는 것 재현하기
    //
    // ★ 이 스텝에서 가장 실전적인 문제입니다.
    //
    // (a) 오래 도는 배치를 실행하십시오. 하루치(389건)는 0.4초라 너무
    //     짧으니, 전체 70,000건을 도는 Job 을 쓰거나 Processor 에
    //     Thread.sleep 을 넣으십시오.
    //
    // (b) 실행 중에 다른 터미널에서 프로세스를 강제 종료하십시오.
    //
    //       ps aux | grep spring-batch5-lab
    //       kill -9 <PID>
    //
    // (c) 메타데이터를 확인하십시오. 무엇이 남아 있습니까?
    //
    //       SELECT JOB_EXECUTION_ID, STATUS, START_TIME, END_TIME
    //       FROM BATCH_JOB_EXECUTION ORDER BY JOB_EXECUTION_ID DESC LIMIT 3;
    //
    //     STATUS:   ____
    //     END_TIME: ____
    //
    // (d) 이제 SettlementScheduler.runNow() 를 호출하십시오.
    //     무슨 일이 벌어집니까? 로그에 무엇이 찍힙니까?
    //
    // (e) ⚠️ 핵심 질문: 이 상태를 방치하면 정산 배치는 언제 다시 돕니까?
    //     중복을 막으려던 장치가 무엇을 막고 있습니까?
    //
    // (f) ZombieCleaner 를 완성하고, @PostConstruct 를 활성화한 뒤
    //     재기동해 좀비가 정리되는 것을 확인하십시오.
    //
    // (g) ⚠️ 마지막 질문: ZOMBIE_THRESHOLD_HOURS 를 얼마로 잡아야 합니까?
    //     너무 짧게 잡으면 어떤 사고가 납니까?
    // =====================================================================

    // (c) 남아 있는 것
    // 여기에 작성:
    //

    // (d) runNow() 호출 시
    // 여기에 작성:
    //

    // (e) 정산 배치는 언제 다시 도는가
    // 여기에 작성:
    //

    // (g) 임계 시간을 얼마로, 너무 짧으면?
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 3. DB 락 기반 3차 방어 구현하고 동시 실행으로 검증하기
    //
    // (a) Practice.JobLockService 를 참고해 락 획득/해제를 구현하고,
    //     Job 실행 전후에 붙이십시오.
    //
    // (b) ⚠️ 두 프로세스를 **동시에** 띄워야 합니다.
    //     순차로 실행하면 락이 이미 해제되어 있어 경쟁 상태가
    //     재현되지 않습니다.
    //
    //       java -jar app.jar ... date=2025-03-02 &
    //       java -jar app.jar ... date=2025-03-02 &
    //       wait
    //
    //     또는 터미널 두 개를 준비해 동시에 엔터를 치십시오.
    //
    // (c) 두 프로세스의 로그를 비교하십시오. 어느 쪽이 락을 얻었습니까?
    //     진 쪽은 어떻게 종료됐습니까?
    //
    // (d) ⚠️ 까다로운 질문: 락 획득에 실패한 프로세스의 종료 코드를
    //     무엇으로 해야 합니까?
    //       - 1(실패)로 하면 무슨 문제가 생깁니까?
    //       - 0(성공)으로 하면 무슨 문제가 생깁니까?
    //     여러분의 답과 근거를 적으십시오.
    //
    // (e) 락 해제가 실패하면(프로세스가 죽으면) 어떻게 됩니까?
    //     이 위험을 어떻게 줄일 수 있습니까?
    // =====================================================================

    // (c) 로그 비교
    // 여기에 작성:
    //

    // (d) 락 실패 시 종료 코드
    // 여기에 작성:
    //

    // (e) 락 해제 실패 대비
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 4. "최근 7일간 재시도가 있었던 날" 찾는 쿼리
    //
    // (a) 하루에 dailySettlementJob 이 2회 이상 실행된 날을 찾으십시오.
    //
    // (b) 거기서 한 걸음 더 나아가십시오. 단순히 2회 이상이 아니라,
    //     **첫 실행이 실패하고 나중 실행이 성공한 날**을 찾으십시오.
    //     이것이 "조용히 넘어간 날"입니다.
    //
    // (c) ⚠️ 왜 이 쿼리가 중요합니까?
    //     재시도해서 결국 성공했다면 알림이 안 갔을 수 있습니다.
    //     그런 날이 반복되면 무엇을 의미합니까?
    // =====================================================================

    public static final String PROBLEM4_QUERY = """
            -- 여기에 작성:
            --
            """;

    // (c) 왜 중요한가
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 5. "배치가 아예 돌지 않았음"을 감지하는 알림 규칙
    //
    // ★ 정답이 하나가 아닙니다. 판단을 정당화하는 것이 핵심입니다.
    //
    // (a) dailySettlementJob 은 매일 새벽 2시에 돕니다.
    //     "돌아야 하는데 안 돌았다"를 감지하는 PromQL 을 작성하십시오.
    //     사용할 지표: spring_batch_job_seconds_count{name, status}
    //
    // (b) 시간 창(threshold)을 몇 시간으로 잡겠습니까?
    //     24시간? 26시간? 30시간? 그 근거를 대십시오.
    //
    // (c) ⚠️ 트레이드오프:
    //       - 너무 짧게 잡으면 무슨 문제가 생깁니까?
    //       - 너무 길게 잡으면 무슨 문제가 생깁니까?
    //
    // (d) absent() 를 쓰는 대안도 있습니다. 두 방식을 비교하십시오.
    //     어떤 상황에서 absent() 가 더 낫습니까?
    //
    // (e) ⚠️ 마지막 질문: "실패 알림"과 "안 돌았음 알림" 중
    //     어느 쪽이 더 중요합니까? 왜입니까?
    // =====================================================================

    public static final String PROBLEM5_ALERT_RULE = """
            groups:
              - name: batch
                rules:
                  # 여기에 작성:
                  #
            """;

    // (b) 시간 창과 근거
    // 여기에 작성:
    //

    // (c) 트레이드오프
    // 여기에 작성:
    //

    // (e) 어느 알림이 더 중요한가
    // 여기에 작성:
    //

    // =====================================================================
    // 문제 6. 종합 실습에 검증 Step 추가하기
    //
    // ★ 이 코스의 마지막 문제입니다.
    //
    // 요구사항: 정산이 끝난 뒤, 정산 결과가 올바른지 검증하는 Step 을
    //           추가하십시오.
    //
    //   (1) orders 의 해당 날짜 COMPLETED 건수와
    //       settlement 의 해당 날짜 정산 건수를 비교합니다.
    //   (2) 두 숫자가 다르면 Job 을 실패시킵니다.
    //       단, skip 된 건수는 정상적인 차이로 인정합니다.
    //   (3) 검증 실패 시 어떤 주문이 누락됐는지 로그에 남깁니다.
    //
    // (a) 검증 Step 을 구현하십시오.
    //
    // (b) ⚠️ 함정 1 (Step 01): 검증 실패를 알리려고
    //     contribution.setExitStatus(ExitStatus.FAILED) 를 하면
    //     어떻게 됩니까? Job 이 실패합니까?
    //
    // (c) ⚠️ 함정 2 (Step 10): 검증 Step 을 흐름에 붙일 때
    //     .on("FAILED").to(...).end() 로 처리하면 어떻게 됩니까?
    //     실패가 제대로 전파됩니까?
    //
    // (d) 위 두 함정을 피해 Job 흐름을 완성하십시오.
    //
    // (e) 일부러 settlement 에서 몇 행을 지우고 검증이 실패하는지
    //     확인하십시오.
    //
    //       DELETE FROM settlement WHERE settle_date='2025-03-01' LIMIT 5;
    //
    //     그리고 java -jar ... 의 종료 코드가 1인지 확인하십시오.
    // =====================================================================

    public static Step problem6VerifyStep(JobRepository jobRepository,
                                          PlatformTransactionManager txManager,
                                          DataSource dataSource,
                                          String date) {
        return new StepBuilder("verifyStep", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    // 여기에 작성:
                    //
                    return org.springframework.batch.repeat.RepeatStatus.FINISHED;
                }, txManager)
                .build();
    }

    // (b) setExitStatus(FAILED) 로 하면?
    // 여기에 작성:
    //

    // (c) .on("FAILED").to(...).end() 로 하면?
    // 여기에 작성:
    //

    // (d) 올바른 Job 흐름
    // 여기에 작성:
    //

    // -- 검증:
    // SELECT (SELECT COUNT(*) FROM orders
    //          WHERE status='COMPLETED' AND DATE(ordered_at)='2025-03-01') AS 대상,
    //        (SELECT COUNT(*) FROM settlement
    //          WHERE settle_date='2025-03-01')                             AS 정산;
}

package com.example.batch.step14;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

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

/**
 * Step 14 — 연습문제 정답과 해설
 *
 * 이 코스의 마지막 파일입니다. 문제를 직접 풀어 본 뒤에 여십시오.
 */
public class Solution {

    // =====================================================================
    // 정답 1. 종료 코드 확인
    // =====================================================================

    /*
     * (a) 성공 시 → **0**
     *
     *     java -jar app.jar --spring.batch.job.enabled=true \
     *       --spring.batch.job.name=dailySettlementJob date=2025-03-01
     *     echo $?
     *     → 0
     *
     * (b) 실패 시 (SpringApplication.exit 감싸기 있음) → **1**
     *
     *     같은 명령을 한 번 더 실행하면
     *     JobInstanceAlreadyCompleteException 이 나고
     *     → 1
     *
     * (c) 실패 시 (감싸기 제거) → **0**
     *
     *     main 을 이렇게 바꾸면:
     *       public static void main(String[] args) {
     *           SpringApplication.run(BatchLabApplication.class, args);
     *       }
     *
     *     Job 이 FAILED 로 끝나도 JVM 은 정상 종료로 간주해 0 을 반환합니다.
     *
     * (d) ⚠️ 왜 위험한가
     *
     *   **파이프라인이 실패를 인지하지 못합니다.**
     *
     *     - 크론: 종료 코드 0 이면 성공으로 보고 메일도 안 보냅니다.
     *     - 에어플로우: 태스크가 초록불로 뜹니다. 다음 태스크가 진행됩니다.
     *     - 쿠버네티스 Job: Succeeded 로 마킹되고 재시도하지 않습니다.
     *     - CI/CD: 배포 검증 단계가 통과합니다.
     *
     *   정산이 실패했는데 **모든 계기판이 초록색**입니다.
     *
     *   그리고 이것이 로그를 봐도 발견되지 않는 이유:
     *   **로그에는 정상적으로 FAILED 가 찍혀 있습니다.**
     *
     *     INFO ... Job: [SimpleJob: [name=dailySettlementJob]] completed
     *              with ... the following status: [FAILED]
     *
     *   로그는 진실을 말하고 있습니다. 문제는 아무도 로그를 안 본다는
     *   것입니다. 사람들은 파이프라인의 초록불을 봅니다. 그리고
     *   파이프라인은 종료 코드만 봅니다.
     *
     *   즉 **로그와 파이프라인이 서로 다른 이야기를 하고 있고, 사람은
     *   틀린 쪽을 봅니다.**
     *
     *   ⚠️ 확인 방법은 딱 하나입니다.
     *      **일부러 실패시켜 보고 echo $? 를 찍어 보십시오.**
     *      새 배치를 운영에 올리기 전 반드시 한 번 하십시오.
     *      이것이 배포 전 체크리스트에서 가장 자주 빠지는 항목입니다.
     */

    // =====================================================================
    // 정답 2. 좀비 실행 — 중복을 막으려던 장치가 실행을 막는다
    // =====================================================================

    /*
     * (c) kill -9 후 남아 있는 것
     *
     *   +------------------+---------+---------------------+----------+
     *   | JOB_EXECUTION_ID | STATUS  | START_TIME          | END_TIME |
     *   +------------------+---------+---------------------+----------+
     *   |               52 | STARTED | 2025-07-20 14:31:07 | NULL     |
     *   +------------------+---------+---------------------+----------+
     *
     *   STATUS  = 'STARTED'
     *   END_TIME = NULL
     *
     *   프로세스가 SIGKILL 로 죽었으므로 셧다운 훅도, 예외 처리도,
     *   afterJob 리스너도 실행되지 않았습니다. Spring Batch 는
     *   "시작했다"까지만 기록하고 그 뒤를 쓰지 못했습니다.
     *
     *   ⚠️ 그리고 **아무도 이 행을 정리해 주지 않습니다.**
     *      다음에 애플리케이션이 떠도, 다음 날이 되어도 그대로입니다.
     *      영원히 STARTED 로 남습니다.
     *
     * (d) runNow() 를 호출하면
     *
     *   WARN c.e.b.step14.SettlementScheduler : 이전 정산 배치가 아직
     *        실행 중입니다. 이번 실행을 건너뜁니다. running=[JobExecution:
     *        id=52, version=1, startTime=2025-07-20T14:31:07, endTime=null,
     *        ...status=STARTED...]
     *
     *   2차 방어가 작동했습니다. findRunningJobExecutions 가 52번을
     *   "실행 중"으로 보고했기 때문입니다.
     *
     *   그런데 52번은 이미 죽은 프로세스입니다. 실행 중이 아닙니다.
     *
     * (e) ⚠️ 정산 배치는 **영원히 다시 돌지 않습니다**
     *
     *   내일도, 모레도, 사람이 손으로 메타데이터를 고치기 전까지
     *   매일 새벽 2시에 "이전 배치가 실행 중"이라며 건너뜁니다.
     *
     *   **중복 실행을 막으려던 장치가 실행 자체를 막고 있습니다.**
     *
     *   이것이 이 스텝에서 가장 실전적인 함정인 이유:
     *     - 에러가 안 납니다. WARN 로그 한 줄이 전부입니다.
     *     - 배치는 "정상적으로" 건너뛰기를 수행합니다.
     *     - 실패 알림이 안 갑니다. 실패한 게 아니라 안 돈 것이니까요.
     *     - 그래서 **14-7 의 "안 돌았음 알림"이 없으면 며칠씩 모릅니다.**
     *
     *   실제로 정산이 3일 밀린 뒤 회계팀이 발견하는 식으로 드러납니다.
     */

    /** (f) 좀비 정리 SQL. */
    public static final String CLEAN_ZOMBIES_SQL = """
            -- 1단계: 좀비 탐지 (먼저 눈으로 확인하십시오)
            SELECT je.JOB_EXECUTION_ID, ji.JOB_NAME, je.START_TIME,
                   TIMESTAMPDIFF(HOUR, je.START_TIME, NOW()) AS hours_ago
            FROM BATCH_JOB_EXECUTION je
            JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
            WHERE je.STATUS IN ('STARTED','STARTING')
              AND je.END_TIME IS NULL
              AND je.START_TIME < NOW() - INTERVAL 6 HOUR;

            -- 2단계: 정리
            UPDATE BATCH_JOB_EXECUTION
               SET STATUS = 'FAILED', EXIT_CODE = 'FAILED',
                   END_TIME = NOW(), LAST_UPDATED = NOW(),
                   EXIT_MESSAGE = '좀비 실행 수동 정리'
             WHERE JOB_EXECUTION_ID = ?;

            -- 3단계: StepExecution 도 함께 정리해야 합니다.
            --        이걸 빼먹으면 재시작 시 Step 상태가 꼬입니다.
            UPDATE BATCH_STEP_EXECUTION
               SET STATUS = 'FAILED', EXIT_CODE = 'FAILED',
                   END_TIME = NOW(), LAST_UPDATED = NOW()
             WHERE JOB_EXECUTION_ID = ? AND END_TIME IS NULL;
            """;

    /*
     * (g) ⚠️ ZOMBIE_THRESHOLD_HOURS 를 얼마로 잡을 것인가
     *
     *   **배치의 최대 실행 시간보다 넉넉히** 잡아야 합니다.
     *
     *   너무 짧게 잡으면 무슨 일이 벌어지는가:
     *
     *     정상적으로 3시간째 돌고 있는 배치가 있는데
     *     임계값을 2시간으로 잡아 두면,
     *     → ZombieCleaner 가 그 배치를 좀비로 오인합니다.
     *     → 메타데이터를 FAILED 로 바꿉니다.
     *     → 그런데 **프로세스는 여전히 살아서 돌고 있습니다.**
     *     → 배치가 끝나면 자기 상태를 COMPLETED 로 쓰려 하는데
     *       메타데이터가 이미 바뀌어 있어 낙관적 락 예외가 납니다.
     *
     *         org.springframework.dao.OptimisticLockingFailureException:
     *           Attempt to update step execution id=... with wrong version
     *
     *     → 최악의 경우, 다른 인스턴스가 "이제 안 돌고 있네" 하고
     *       **같은 배치를 동시에 시작합니다.** 정산이 두 배가 됩니다.
     *
     *   즉 **좀비 정리를 잘못하면 그게 중복 실행의 원인이 됩니다.**
     *   막으려던 것을 스스로 만들어 내는 셈입니다.
     *
     *   실무 지침:
     *     - 임계값 = 배치 최대 실행 시간 × 2 + 여유
     *       (정산이 최대 1시간이면 6시간 정도)
     *     - 자동 정리는 **기동 시점에만** 하십시오. 주기적으로 돌리면
     *       실행 중인 배치와 부딪힐 위험이 커집니다.
     *     - 더 안전한 방법: 정리 대상을 **자동으로 고치지 말고 알림만**
     *       보내고, 사람이 확인 후 처리하게 하는 것. 좀비는 자주 생기는
     *       일이 아니므로 수동 처리로도 충분합니다.
     *     - 근본 해결책은 **프로세스가 살아 있는지 직접 확인**하는 것입니다.
     *       BATCH_JOB_EXECUTION 에 호스트명/PID 를 남기면(커스텀 컬럼이나
     *       ExecutionContext) 진짜 좀비인지 판정할 수 있습니다.
     */

    // =====================================================================
    // 정답 3. DB 락 기반 3차 방어
    // =====================================================================

    /*
     * (c) 두 프로세스의 로그
     *
     *   [서버 A]
     *   INFO c.e.b.step14.JobLockService : 락 획득 성공: dailySettlementJob/2025-03-02
     *   INFO o.s.b.c.l.s.TaskExecutorJobLauncher : Job: [FlowJob: [name=dailySettlementJob]] launched ...
     *   INFO o.s.batch.core.step.AbstractStep : Step: [dailySettlementStep] executed in 408ms
     *   INFO o.s.b.c.l.s.TaskExecutorJobLauncher : ... status: [COMPLETED] in 461ms
     *
     *   [서버 B]
     *   WARN c.e.b.step14.JobLockService : 락 획득 실패 — 다른 인스턴스가 실행 중입니다. 종료합니다.
     *
     *   B 는 Job 을 아예 시작하지 않고 종료했습니다.
     *
     *   ⚠️ 여기서 중요한 것: **1차 방어(JobInstance)는 이 경우를 못
     *      막았을 것입니다.** 두 프로세스가 동시에 시작하면 둘 다
     *      "기존 JobInstance 없음"을 확인하고 둘 다 진행합니다.
     *      DB 의 PK 제약만이 원자적으로 하나를 튕겨냅니다.
     *
     * (d) ⚠️ 락 실패 시 종료 코드 — 까다로운 질문
     *
     *   **1(실패)로 하면:**
     *     크론이 실패로 인지하고 알림을 보냅니다. 그런데 중복 실행을
     *     막은 것은 **정상 동작**입니다. 알림이 갈 이유가 없습니다.
     *     매일 오탐 알림이 오면 사람들이 알림을 무시하게 되고,
     *     결국 진짜 실패도 놓칩니다. (알림 피로)
     *
     *   **0(성공)으로 하면:**
     *     파이프라인이 "배치가 돌았다"고 인식합니다. 그런데 안 돌았습니다.
     *     14-7 의 "안 돌았음 알림"도 무력화됩니다. 종료 코드만 보면
     *     성공이니까요.
     *
     *   **권장: 별도 종료 코드(예: 3)를 쓰고 파이프라인에서 구분합니다.**
     *
     *     @Component
     *     public class LockExitCodeGenerator implements ExitCodeGenerator {
     *         public int getExitCode() { return lockAcquired ? 0 : 3; }
     *     }
     *
     *   그리고 크론/에어플로우 쪽에서:
     *     - 0 → 성공
     *     - 3 → 스킵됨. 알림 없음. 단, **연속 3회 이상이면 알림**
     *           (락이 안 풀리고 있다는 뜻)
     *     - 그 외 → 실패. 알림.
     *
     *   "연속 3회 이상이면 알림"이 중요합니다. 락 해제 실패로 인한
     *   교착을 잡아 주기 때문입니다.
     *
     * (e) 락 해제 실패 대비
     *
     *   프로세스가 죽으면 finally 블록도 실행되지 않아 락이 남습니다.
     *   좀비 실행과 똑같은 문제입니다.
     *
     *   대비책 세 가지:
     *
     *   ① **TTL 을 두십시오.**
     *      락 테이블에 acquired_at 이 있으므로, 획득 시 오래된 락을
     *      먼저 지웁니다.
     *        DELETE FROM batch_job_lock
     *         WHERE job_name = ? AND acquired_at < NOW() - INTERVAL 6 HOUR;
     *      그 다음 INSERT 를 시도합니다. 원자성을 유지하려면 이 둘을
     *      한 트랜잭션에 넣으십시오.
     *
     *   ② **holder 컬럼을 활용하십시오.**
     *      호스트명이 기록되므로, 그 호스트가 살아 있는지 확인할 수
     *      있습니다. 운영자가 조사할 때 결정적인 단서가 됩니다.
     *
     *   ③ **락 대신 Quartz 클러스터링을 쓰십시오.**
     *      isClustered: true 로 두면 Quartz 가 자체 DB 락으로 하나만
     *      실행하도록 보장하고, **TTL 관리도 알아서 합니다.**
     *      직접 만든 락보다 검증된 구현을 쓰는 편이 낫습니다.
     *
     *   ⚠️ 일반 원칙: **분산 락에는 반드시 만료 시간이 있어야 합니다.**
     *      만료 없는 락은 언젠가 반드시 교착을 만듭니다.
     */

    // =====================================================================
    // 정답 4. 재시도가 있었던 날 찾기
    // =====================================================================

    /** (a) 단순 버전 — 하루에 2회 이상 실행된 날. */
    public static final String RETRIED_DAYS_SIMPLE = """
            SELECT DATE(je.START_TIME) AS d, COUNT(*) AS runs
            FROM BATCH_JOB_EXECUTION je
            JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
            WHERE ji.JOB_NAME = 'dailySettlementJob'
              AND je.START_TIME >= NOW() - INTERVAL 7 DAY
            GROUP BY DATE(je.START_TIME)
            HAVING runs > 1
            ORDER BY d DESC;
            """;

    /**
     * (b) 정밀 버전 — "조용히 넘어간 날".
     *
     * 첫 실행이 실패하고 나중 실행이 성공한 날만 골라냅니다.
     * 단순히 2회 이상인 것과 다릅니다. 운영자가 파라미터를 바꿔
     * 두 번 돌린 경우(둘 다 성공)는 제외됩니다.
     */
    public static final String SILENTLY_RECOVERED_DAYS = """
            SELECT DATE(je.START_TIME)                          AS d,
                   COUNT(*)                                     AS runs,
                   SUM(je.STATUS = 'FAILED')                    AS failures,
                   SUM(je.STATUS = 'COMPLETED')                 AS successes,
                   MIN(CASE WHEN je.STATUS = 'FAILED'
                            THEN LEFT(je.EXIT_MESSAGE, 60) END) AS first_error
            FROM BATCH_JOB_EXECUTION je
            JOIN BATCH_JOB_INSTANCE ji USING (JOB_INSTANCE_ID)
            WHERE ji.JOB_NAME = 'dailySettlementJob'
              AND je.START_TIME >= NOW() - INTERVAL 7 DAY
            GROUP BY DATE(je.START_TIME)
            HAVING failures > 0 AND successes > 0
            ORDER BY d DESC;
            """;

    /*
     * 결과 예시:
     *
     * +------------+------+----------+-----------+----------------------------------+
     * | d          | runs | failures | successes | first_error                      |
     * +------------+------+----------+-----------+----------------------------------+
     * | 2025-07-19 |    2 |        1 |         1 | org.springframework.dao.Deadlock |
     * | 2025-07-16 |    3 |        2 |         1 | java.lang.IllegalArgumentExcep   |
     * +------------+------+----------+-----------+----------------------------------+
     *
     * (c) ⚠️ 왜 이 쿼리가 중요한가
     *
     *   **재시도해서 결국 성공한 날은 알림이 안 갔을 가능성이 큽니다.**
     *
     *   에어플로우나 쿠버네티스의 자동 재시도(backoffLimit)를 켜 두면,
     *   1회차가 실패해도 2회차가 성공하면 태스크는 초록불입니다.
     *   사람은 아무것도 모릅니다.
     *
     *   하루 이틀이면 "일시적 데드락이었나 보다" 하고 넘어갈 수 있습니다.
     *   그런데 이런 날이 **반복되면** 그건 다른 이야기입니다.
     *
     *     - 데드락이 반복 → 다른 배치와 락 순서가 충돌하고 있음
     *     - 타임아웃이 반복 → 데이터가 늘어 배치가 한계에 근접
     *     - 특정 요일에 집중 → 주간 배치와 겹치는 스케줄 문제
     *
     *   전부 **재시도로 덮이지만 근본 원인이 있는** 경우입니다.
     *   그리고 언젠가는 재시도로도 안 되는 날이 옵니다. 대개
     *   데이터가 가장 많은 날, 즉 가장 중요한 날입니다.
     *
     *   이 쿼리를 주간 리포트에 넣으십시오. "조용한 재시도"의 추세가
     *   장애의 선행 지표입니다.
     */

    // =====================================================================
    // 정답 5. "안 돌았음" 알림 규칙
    // =====================================================================

    /*
     * (a)(b) PromQL 과 시간 창 → **26시간을 권장합니다**
     *
     *   groups:
     *     - name: batch
     *       rules:
     *         - alert: SettlementJobMissing
     *           expr: |
     *             time() - max(
     *               spring_batch_job_seconds_count{
     *                 name="dailySettlementJob", status="COMPLETED"
     *               }
     *             ) > 93600
     *           for: 10m
     *           labels:
     *             severity: critical
     *           annotations:
     *             summary: "정산 배치가 26시간째 성공하지 않았습니다"
     *
     *   93600초 = 26시간
     *
     *   근거:
     *     - 배치는 매일 새벽 2시에 돕니다 → 정상 간격은 24시간
     *     - 배치 자체가 최대 1시간 걸릴 수 있음 → +1시간
     *     - 스케줄러 지연, 락 대기, 재시도 여유 → +1시간
     *     → 26시간
     *
     * (c) ⚠️ 트레이드오프
     *
     *   **너무 짧게 잡으면 (예: 24시간)**
     *     배치가 조금만 늦어도 알림이 갑니다. 새벽 2시에 도는 배치가
     *     2시 30분에 돌면 그날은 24.5시간 간격이 되어 오탐입니다.
     *     매주 두세 번씩 오탐이 오면 사람들이 알림 채널을 음소거합니다.
     *     그리고 진짜 장애를 놓칩니다. **알림 피로가 알림 부재보다
     *     위험합니다.**
     *
     *   **너무 길게 잡으면 (예: 50시간)**
     *     이틀치 정산을 놓치고 나서야 알림이 옵니다. 정산 배치라면
     *     이미 회계 마감에 영향을 준 뒤입니다.
     *     감지 지연 자체가 손실입니다.
     *
     *   26시간이면 "새벽 2시에 안 돌았으면 다음 날 새벽 4시에 알림"
     *   입니다. 하루치만 놓치고 잡히므로 복구 가능한 범위입니다.
     *
     * (d) absent() 대안 비교
     *
     *   - alert: SettlementJobNeverRan
     *     expr: absent(spring_batch_job_seconds_count{name="dailySettlementJob"})
     *     for: 1h
     *
     *   | | time() - max(...) | absent(...) |
     *   |---|---|---|
     *   | 감지 대상 | "최근에 성공한 적 없음" | "지표 자체가 없음" |
     *   | 배치가 계속 실패 중 | **감지함** | 감지 못 함(지표는 있으니까) |
     *   | 배치 앱이 아예 안 뜸 | 감지함 | **감지함** |
     *   | 새로 배포한 직후 | 오탐 위험 | 오탐 위험 |
     *
     *   **둘 다 거십시오.** 서로 다른 실패 양상을 잡습니다.
     *   absent() 는 "애플리케이션이 아예 없다"(배포 누락, 파드 스케일 0)를
     *   잡고, time()-max() 는 "떠 있는데 안 돈다"를 잡습니다.
     *
     *   ⚠️ 그리고 일회성 CLI 배치라면 두 규칙 모두 Pushgateway 가
     *      전제입니다(14-7). pull 방식으로는 30초 만에 죽는 프로세스의
     *      지표를 못 긁습니다.
     *
     * (e) ⚠️ "실패 알림"과 "안 돌았음 알림" 중 무엇이 더 중요한가
     *
     *   **"안 돌았음 알림"이 더 중요합니다.**
     *
     *   실패는 이미 시끄럽습니다.
     *     - 로그에 스택트레이스가 남습니다
     *     - EXIT_MESSAGE 에 원인이 저장됩니다
     *     - 종료 코드가 1입니다
     *     - 메트릭에 status="FAILED" 가 올라갑니다
     *   네 가지 경로 중 하나만 걸려도 발견됩니다.
     *
     *   그런데 "아예 안 돈 것"은 **아무 흔적이 없습니다.**
     *     - 로그가 없습니다 (실행이 없었으니까)
     *     - 메타데이터에 행이 안 생깁니다
     *     - 종료 코드가 없습니다 (프로세스가 없었으니까)
     *     - 메트릭이 안 올라갑니다
     *
     *   **없음을 감지하려면 없음을 감시해야 합니다.** 그리고 이건
     *   의도적으로 규칙을 만들어야만 됩니다.
     *
     *   그리고 이 코스에서 배운 것 중 "안 돌게 만드는" 원인이
     *   얼마나 많았는지 떠올려 보십시오.
     *     - @EnableBatchProcessing 으로 자동설정이 꺼짐 (Step 02)
     *     - spring.batch.job.enabled: false 로 배포 (Step 14)
     *     - 좀비 실행 때문에 2차 방어가 영원히 스킵 (Step 14)
     *     - 스케줄러 풀 크기 1이라 다른 배치에 밀림 (Step 14)
     *     - Quartz misfire 정책으로 조용히 버려짐 (Step 14)
     *
     *   전부 "에러 없이 아무 일도 안 일어나는" 사고입니다.
     *   이 코스의 주제 그 자체입니다.
     */

    // =====================================================================
    // 정답 6. 검증 Step — 이 코스의 마지막 코드
    // =====================================================================

    /** (a) 검증 Step. */
    public static Step verifyStep(JobRepository jobRepository,
                                  PlatformTransactionManager txManager,
                                  DataSource dataSource,
                                  String date) {

        Logger log = LoggerFactory.getLogger("VerifyStep");

        return new StepBuilder("verifyStep", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    JdbcTemplate jdbc = new JdbcTemplate(dataSource);

                    Integer target = jdbc.queryForObject("""
                            SELECT COUNT(*) FROM orders
                            WHERE status = 'COMPLETED' AND DATE(ordered_at) = ?
                            """, Integer.class, date);

                    Integer settled = jdbc.queryForObject("""
                            SELECT COUNT(*) FROM settlement WHERE settle_date = ?
                            """, Integer.class, date);

                    Integer skipped = jdbc.queryForObject("""
                            SELECT COUNT(*) FROM s14_bad_order
                            WHERE DATE(occurred_at) = CURDATE()
                            """, Integer.class);

                    log.info("검증: 대상={}, 정산={}, skip={}", target, settled, skipped);

                    // (2) skip 된 건수는 정상적인 차이로 인정합니다.
                    int expected = target - skipped;
                    if (settled != expected) {
                        // (3) 어떤 주문이 누락됐는지 남깁니다.
                        List<Long> missing = jdbc.queryForList("""
                                SELECT o.order_id FROM orders o
                                LEFT JOIN settlement s ON s.order_id = o.order_id
                                WHERE o.status = 'COMPLETED'
                                  AND DATE(o.ordered_at) = ?
                                  AND s.order_id IS NULL
                                LIMIT 20
                                """, Long.class, date);
                        log.error("정산 누락 감지! 기대={}, 실제={}, 누락 예시={}",
                                expected, settled, missing);

                        // ⚠️ (b) 의 핵심: ExitStatus 가 아니라 **예외를 던져야** 합니다.
                        throw new IllegalStateException(
                                "정산 검증 실패: 기대=%d, 실제=%d".formatted(expected, settled));
                    }

                    log.info("검증 통과");
                    return RepeatStatus.FINISHED;
                }, txManager)
                .build();
    }

    /*
     * (b) ⚠️ 함정 1 — setExitStatus(ExitStatus.FAILED) 로는 실패하지 않습니다
     *
     *   contribution.setExitStatus(ExitStatus.FAILED) 를 해도
     *   **BatchStatus 는 COMPLETED 로 남습니다.** (Step 01-5)
     *
     *   결과:
     *     - Job 이 COMPLETED 로 기록됩니다
     *     - 종료 코드가 0 입니다
     *     - 크론이 성공으로 인지합니다
     *     - **재실행이 차단됩니다** (JobInstanceAlreadyComplete)
     *
     *   검증에 실패했는데 "성공적으로 완료"로 남고, 심지어 다시 돌릴
     *   수도 없습니다. 최악의 조합입니다.
     *
     *   ExitStatus 는 "다음에 어느 Step 으로 갈지"를 정하는 꼬리표일 뿐
     *   실패 판정 장치가 아닙니다.
     *
     *   **진짜로 실패시키려면 예외를 던져야 합니다.**
     *
     * (c) ⚠️ 함정 2 — .on("FAILED").to(...).end() 는 실패를 은폐합니다
     *
     *   검증 실패 시 알림 Step 을 붙이고 싶어서 이렇게 쓰기 쉽습니다.
     *
     *     .start(dailySettlementStep)
     *     .next(verifyStep)
     *         .on("FAILED").to(alertStep).end()      // ← 위험
     *         .on("*").end()
     *
     *   그러면 **Step 은 FAILED 인데 Job 은 COMPLETED** 가 됩니다.
     *   (Step 10 의 핵심 함정)
     *
     *   .end() 는 "이 흐름을 성공으로 종료한다"는 뜻이기 때문입니다.
     *   결과적으로:
     *     - 종료 코드 0
     *     - 모니터링 침묵
     *     - 재시작 봉쇄
     *
     *   검증 Step 을 붙인 목적이 통째로 사라집니다. 오히려 검증이
     *   없느니만 못합니다 — 검증했다는 착각을 주니까요.
     *
     * (d) 올바른 Job 흐름 — 두 함정을 모두 피합니다
     */

    public static Job dailySettlementJobWithVerify(JobRepository jobRepository,
                                                   Step dailySettlementStep,
                                                   Step reportStep,
                                                   Step verifyStep,
                                                   Step alertStep) {
        return new JobBuilder("dailySettlementJob", jobRepository)
                .start(dailySettlementStep)
                .next(reportStep)
                .next(verifyStep)
                    // 검증 실패 시 알림 Step 을 거치되,
                    // 반드시 .fail() 로 끝내 실패를 전파합니다.
                    .on("FAILED").to(alertStep).on("*").fail()
                .from(verifyStep)
                    .on("*").end()
                .end()
                .build();
    }

    /*
     * 핵심은 `.on("FAILED").to(alertStep).on("*").fail()` 입니다.
     *
     *   - alertStep 은 실행됩니다 (알림은 갑니다)
     *   - 그 뒤 .fail() 이 Job 을 FAILED 로 만듭니다
     *   - 종료 코드 1
     *   - 재시작 가능
     *   - 모니터링 알림 발생
     *
     * (e) 검증 확인
     *
     *   mysql> DELETE FROM settlement WHERE settle_date='2025-03-01' LIMIT 5;
     *   Query OK, 5 rows affected
     *
     *   $ java -jar app.jar ... date=2025-03-01 attempt=verify
     *
     *   INFO  VerifyStep : 검증: 대상=389, 정산=384, skip=0
     *   ERROR VerifyStep : 정산 누락 감지! 기대=389, 실제=384,
     *                      누락 예시=[10, 1010, 2010, 3010, 4010]
     *   ERROR o.s.batch.core.step.AbstractStep : Encountered an error
     *         executing step verifyStep in job dailySettlementJob
     *   java.lang.IllegalStateException: 정산 검증 실패: 기대=389, 실제=384
     *   INFO  o.s.b.c.l.s.TaskExecutorJobLauncher : Job: [FlowJob:
     *         [name=dailySettlementJob]] completed ... status: [FAILED]
     *
     *   $ echo $?
     *   1
     *
     *   ─────────────────────────────────────────────────────────────
     *   이 마지막 문제에서 이 코스의 두 교훈이 합쳐집니다.
     *
     *     Step 01 — 실패시키려면 ExitStatus 가 아니라 예외를 던져라
     *     Step 10 — .end() 로 잡으면 실패가 은폐된다
     *
     *   그리고 그 위에 Step 14 의 교훈이 얹힙니다.
     *
     *     종료 코드 1 을 눈으로 확인하기 전까지,
     *     그 배치가 실패를 제대로 보고한다고 믿지 마십시오.
     *
     *   ─────────────────────────────────────────────────────────────
     *   마지막으로, 이 코스의 한 문장을 다시 남깁니다.
     *
     *     배치가 COMPLETED 로 끝났다는 사실은,
     *     그 배치가 옳게 동작했다는 뜻이 아닙니다.
     *
     *   수고하셨습니다.
     *   ─────────────────────────────────────────────────────────────
     */
}