spring:
  # 1. Spring Batch 프레임워크 핵심 제어 옵션
  batch:
    jdbc:
      # always: 기동 시 테이블이 없으면 자동 생성 (로컬/테스트 환경 추천)
      # never: 테이블을 자동 생성하지 않음 (DDL 권한이 제한된 운영 환경 추천)
      initialize-schema: always 
    job:
      # false: 어플리케이션이 켜질 때 모든 Job이 자동으로 실행되는 것을 방지합니다.
      # 운영 환경에서는 스케줄러(Jenkins, Quartz 등)가 원하는 시점에만 호출해야 하므로 필수 설정입니다.
      enabled: false 
 
  # 2. 멀티 데이터소스(Multi-DataSource) 설정
  datasource:
    primary:
      jdbc-url: jdbc:mysql://localhost:3306/business_db?serverTimezone=Asia/Seoul
      username: main_user
      password: main_password
      driver-class-name: com.mysql.cj.jdbc.Driver
      hikari:
        pool-name: HikariPool-Main
        maximum-pool-size: 10
 
    batch:
      jdbc-url: jdbc:mysql://localhost:3306/batch_meta_db?serverTimezone=Asia/Seoul
      username: batch_user
      password: batch_password
      driver-class-name: com.mysql.cj.jdbc.Driver
      hikari:
        pool-name: HikariPool-Batch
        maximum-pool-size: 5

package com.example.batch.config;
 
import javax.sql.DataSource;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
 
@Configuration
public class BatchDataSourceConfig {
 
    // 1. 비즈니스 로직용 메인 데이터소스
    @Bean
    @Primary // @Autowired 시 기본 채택되는 데이터소스 지정
    @ConfigurationProperties(prefix = "spring.datasource.primary")
    public DataSource primaryDataSource() {
        return DataSourceBuilder.create().build();
    }
 
    // 2. 배치 메타데이터 테이블용 데이터소스
    @Bean(name = "batchDataSource")
    @ConfigurationProperties(prefix = "spring.datasource.batch")
    public DataSource batchDataSource() {
        return DataSourceBuilder.create().build();
    }
 
    // 3. 배치 전용 트랜잭션 매니저 (배치 데이터소스 연동)
    @Bean(name = "batchTransactionManager")
    public PlatformTransactionManager batchTransactionManager(
            @Qualifier("batchDataSource") DataSource batchDataSource) {
        return new DataSourceTransactionManager(batchDataSource);
    }
 
    // 4. 커스텀 JobRepository 설정
    @Bean
    public JobRepository jobRepository(
            @Qualifier("batchDataSource") DataSource batchDataSource,
            @Qualifier("batchTransactionManager") PlatformTransactionManager batchTransactionManager) throws Exception {
        
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(batchDataSource);
        factory.setTransactionManager(batchTransactionManager);
        
        // [성능 및 격리 튜닝]
        // 대량의 배치 메타데이터가 적재될 때 데드락(Deadlock)을 방지하기 위해 격리 수준을 조정합니다.
        // 해당 DB 제품군이 지원하는 최적의 격리 수준을 선택합니다. (기본값: ISOLATION_SERIALIZABLE)
        factory.setIsolationLevelForCreate("ISOLATION_READ_COMMITTED");
        
        // 만약 한 DB 내에서 여러 서비스의 배치가 섞인다면 테이블 접두사를 변경할 수 있습니다. (기본값: BATCH_)
        factory.setTablePrefix("BATCH_"); 
        
        return factory.getObject();
    }
}

package com.example.batch.job;
 
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
 
import java.util.Arrays;
 
@Slf4j
@Configuration
@RequiredArgsConstructor
public class UserSettlementJobConfig {
 
    // 1. Job 구성: JobRepository를 주입받아 메타데이터와 연동합니다.
    @Bean
    public Job userSettlementJob(JobRepository jobRepository, Step settlementStep) {
        return new JobBuilder("userSettlementJob", jobRepository)
                .start(settlementStep)
                .build();
    }
 
    // 2. Step 구성: Chunk 크기(Commit Interval)와 트랜잭션 매니저를 지정합니다.
    @Bean
    @JobScope
    public Step settlementStep(JobRepository jobRepository, 
                               PlatformTransactionManager batchTransactionManager,
                               ItemReader<String> settlementReader,
                               ItemProcessor<String, String> settlementProcessor,
                               ItemWriter<String> settlementWriter) {
        return new StepBuilder("settlementStep", jobRepository)
                .<String, String>chunk(5, batchTransactionManager) // Commit Interval = 5
                .reader(settlementReader)
                .processor(settlementProcessor)
                .writer(settlementWriter)
                .build();
    }
 
    // 3. ItemReader: Late Binding을 통해 JobParameters를 주입받아 활용합니다.
    @Bean
    @StepScope
    public ListItemReader<String> settlementReader(
            @Value("#{jobParameters['requestDate']}") String requestDate) {
        
        log.info("--> [Reader] BATCH_JOB_EXECUTION_PARAMS에서 requestDate 파라미터 읽기: {}", requestDate);
        
        // 실무에서는 여기서 CursorReader나 PagingReader를 사용해 DB를 조회합니다.
        return new ListItemReader<>(Arrays.asList("user1", "user2", "user3", "user4", "user5", "user6"));
    }
 
    // 4. ItemProcessor: 비즈니스 로직 가공 및 필터링
    @Bean
    @StepScope
    public ItemProcessor<String, String> settlementProcessor() {
        return item -> {
            log.info("--> [Processor] 데이터 가공 처리 중: {}", item);
            return item + " 완료";
        };
    }
 
    // 5. ItemWriter: Chunk 단위(5건)로 모인 리스트를 받아 일괄 저장합니다.
    @Bean
    @StepScope
    public ItemWriter<String> settlementWriter() {
        return chunk -> {
            log.info("--> [Writer] Chunk 단위 일괄 등록 착수 (Size: {})", chunk.size());
            for (String item : chunk) {
                log.info("--> [Writer] DB 영속화 진행: {}", item);
            }
            // 이 블록이 예외 없이 끝나면 batchTransactionManager에 의해 부모 트랜잭션이 커밋됩니다.
        };
    }
}

package com.example.batch.listener;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.AfterStep;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.stereotype.Component;
 
@Slf4j
@Component
public class SettlementStepListener {
 
    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        // BATCH_STEP_EXECUTION_CONTEXT 조회 및 데이터 적재
        ExecutionContext stepContext = stepExecution.getExecutionContext();
        
        // 이전 실행 실패로 인해 기존 컨텍스트 데이터가 남아있는지 확인 (Restart 체크)
        if(stepContext.containsKey("lastProcessedIndex")) {
            int lastIndex = stepContext.getInt("lastProcessedIndex");
            log.info("--> [Restart 감지] 과거 실패 지점 Index 존재: {}", lastIndex);
        } else {
            stepContext.putInt("lastProcessedIndex", 0); // 초기값 세팅
            log.info("--> [최초 실행] ExecutionContext 내 lastProcessedIndex 초기화 완료");
        }
    }
 
    @AfterStep
    public ExitStatus afterStep(StepExecution stepExecution) {
        ExecutionContext stepContext = stepExecution.getExecutionContext();
        
        // Step 종료 직전 상태값 업데이트 -> BATCH_STEP_EXECUTION_CONTEXT 테이블에 직렬화되어 저장됨
        stepContext.putInt("lastProcessedIndex", 9999); 
        
        log.info("--> [Step 완료] 최종 누적 메트릭 - Read Count: {}, Write Count: {}", 
                stepExecution.getReadCount(), stepExecution.getWriteCount());
        
        return stepExecution.getExitStatus();
    }
}