섹션 6. 스프링 배치 실행 - Flow

[0JUUU]

• [x] FlowJob - 개념 및 API 소개
• [x] FlowJob - start() / next()
• [x] Transition - 배치상태 유형 (BatchStatus / ExitStatus / FlowExecutionStatus)
• [x] Transition - on() / to() / stop(), fail(), end(), stopAndRestart()
• [x] 사용자 정의 ExitStatus
• [x] JobExecutionDecider
• [x] FlowJob 아키텍처
• [x] SimpleFlow - 개념 및 API 소개
• [x] SimpleFlow 예제
• [x] SimpleFlow 아키텍처
• [x] FlowStep
• [x] @JobScope / @StepScope - 기본개념 및 설정
• [x] @JobScope / @StepScope 아키텍처

[0JUUU]

FlowJob - 개념 및 API 소개

• Step을 순차적으로 구성하는 것이 아닌 특정한 상태에 따라 흐름을 전환하도록 구성 ➡️ FlowJobBuilder에 의해 생성

  • Step이 실패하더라도 Job은 실패로 끝나지 않아야할 경우
  • Step이 성공했을 때 다음에 실행해야 할 Step을 구분해야할 경우
  • 특정 Step은 전혀 실행되지 않게 구성해야하는 경우

• Flow : 흐름 정의하는 역할 (e.g. start(), from(), next())

• Transition : 조건에 따라 흐름을 전환시키는 역할 (e.g. on(), to(), stop(), fail(), end(), stopAndRestart())

    @Bean
    public Job batchJob() {
        return jobBuilderFactory.get("batchJob")
                                .start(step1())
                                    .on("COMPLETED").to(step2())
                                .from(step1())
                                    .on("FAILED").to(step3())
                                .end()
                                .build();
    }

FlowJob - 성공

FlowJob - Step1 실패

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

• 선행된 Step이 실패하여도 이후 실행된 Step은 완료할 수 있다 ➡️ Job도 Complete 상태
• 반드시 Job을 성공해야할 때는 FlowJob을 사용하는 것이 좋다.

[0JUUU]

FlowJob - start() / next()

start()

• Flow, Step 모두가 파라미터로 넘어갈 수 있다.
• Flow가 오게 되면 jobFlowBuilder 반환
• Step이 오면 SimpleJobBuilder 반환
  • SimpleJobBuilder도 on을 제공 ➡️ start에 step은 인자로 넣고 뒤에서 on을 사용하면 jobFlowBuilder가 반환됨

next()

• Step, Flow, JobExecutionDecider 타입을 받을 수 있음

의도적인 에러 발생

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

• start / next : on,from,... 등과는 다르게 Step 중 하나라도 실패할 경우 전체 Job이 실패하게 된다.

[0JUUU]

Transition - 배치상태 유형 (BatchStatus / ExitStatus / FlowExecutionStatus)

BatchStatus (Enum 형태로 정의)

• JobExecution과 StepExecution 속성 ➡️ Job과 Step의 종료 후 최종 결과 상태가 무엇인지 정의
• SimpleJob
  • 마지막 Step의 BatchStatus 값을 Job의 최종 BatchStatus값으로 반영
  • Step이 실패할 경우 해당 Step이 마지막 Step이 됨
• FlowJob
  • Flow 내 Step의 ExitStatus값을 FlowExecutionStatus 값으로 저장
  • 마지막 Flow의 FlowExecutionStatus 값을 Job의 최종 BatchStatus 값으로 반영
• COMPLETED, STARTING, STARTED, STOPPING, STOPPED, FAILED, ABANDONED, UNKNOWN
• ABANDONED : 처리를 완료했지만 성공하지 못한 단계와 재시작 시 건너뛰어야하는 단계

ExitStatus

• JobExecution과 StepExecution의 속성 ➡️ Job과 Step의 실행 후 어떤 상태로 종료되었는지 정의
• 기본적으로 ExitStatus와 BatchStatus는 동일한 값으로 설정 (물론, 다르게 설정할 수 있는 방법 존재)
• SimpleJob
  • 마지막 Step의 ExitStatus 값을 Job의 최종 ExitStatus값으로 반영
• FlowJob
  • Flow 내 Step의 ExitStatus값을 FlowExecutionStatus 값으로 저장
  • 마지막 Flow의 FlowExecutionStatus 값을 Job의 최종 ExitStatus 값으로 반영
• UNKNOWN, EXECUTING, COMPLETED, NOOP, FAILED, STOPPED

FlowExecutionStatus

• FlowExecution 속성 ➡️ Flow 실행 후 최종 결과 상태가 무엇인지 정의
• Flow 내 Step이 실행되고 나서 ExitStatus값을 FlowExecutionStatus 값으로 저장
• FlowJob의 배치 결과 상태에 관여
• COMPLETED, STOPPED, FAILED, UNKNOWN

@RequiredArgsConstructor
@Configuration
public class TransitionConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job batchJob() {
        return jobBuilderFactory.get("batchJob")
                                .start(step1())
                                .next(step2())
                                .build();
    }
@Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                 .tasklet(new Tasklet() {
                                     @Override
                                     public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                                         System.out.println(">> step1 has executed");
                                         return RepeatStatus.FINISHED;
                                     }
                                 }).build();
    }

    @Bean
    public Step step2() {
        return stepBuilderFactory.get("step2")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step2 has executed");
                                     contribution.setExitStatus(ExitStatus.FAILED);
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }
}

• Step2의 ExitStatus를 FAILED로 설정
• Job은 마지막에 실행된 Step의 ExitStatus를 가지고 가기에 얘도 FAILED

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

FlowJob

[0JUUU]

Transition - on() / to() / stop(), fail(), end(), stopAndRestart()

Transition

• Flow 내 Step의 조건부 전환(전이) 정의
• Job의 API 설정에서 on(String Pattern) 호출 ➡️ TransitionBuilder 반환 ➡️ Transition Flow 구성 가능
• Step의 종료상태(ExitStatus)가 어떤 pattern과도 매칭 ❌ ➡️ 스프링 배치에서 예외 발생 & Job 실패
• 구체적인 것부터 그렇지 않은 순서로 적용

Transition - API

• on(String pattern)
  • Step의 실행 결과로 돌려받는 종료상태(ExitStatus)와 매칭하는 패턴 스키마. BatchStatus와 매칭 ❌
  • pattern과 ExitStatus 매칭 ➡️ 다음으로 실행할 Step 지정 가능
  • 특수문자는 두 가지만 허용
    • "*" : 0개 이상의 문자와 매칭, 모든 ExitStatus와 매칭
    • "?" : 정확히 1개의 문자와 매칭
    • e.g. "c*t" : "cat", "count" 모두 매칭 / "c?t" : "cat"만 매칭
• to()
  • 다음으로 실행할 단계 지정
• from()
  • 이전 단계에서 정의한 Transition을 새롭게 추가 정의

Job을 중단하거나 종료하는 Transition API

• Flow 실행 ➡️ FlowExecutionStatus에 상태값 저장 & 최종적으로 Job의 BatchStatus와 ExitStatus 반영
• Step의 BatchStatus, ExitStatus에는 아무런 영향 ❌, 오로지 Job의 상태만을 변경
• stop()
  • FlowExecutionStatus가 STOPPED 상태로 종료되는 transition
  • Job의 BatchStatus와 ExitStatus가 STOPPED로 종료
• fail()
  • FlowExecutionStatus가 FAILED 상태로 종료되는 transition
  • Job의 BatchStatus와 ExitStatus가 FAILED로 종료
• end()
  • FlowExecutionStatus가 COMPLETED 상태로 종료되는 transition
  • Job의 BatchStatus와 ExitStatus가 COMPLETED로 종료
  • Step의 ExitStatus가 FAILED이더라도 Job의 BatchStatus가 COMPLETED로 종료하도록 할 수 있으며 이 떄 Job의 재시작은 불가능
• stopAndRestart()
  • stop() 과 기본 흐름은 동일
  • 특정 step에서 작업을 중단하도록 설정하면 중단 이전의 Step만 COMPLETED 저장되고 이후 step은 실행되지 않고 STOPPED 상태로 Job 종료
  • Job이 다시 실행됐을 때 실행해야할 step을 restart 인자로 넘기면 이전에 COMPLETED로 저장된 step은 건너뛰고 중단 이후 step부터 시작

    @Bean
    public Job batchJob() {
        return this.jobBuilderFactory.get("batchJob")
                                     .start(step1())
                                         .on("FAILED")
                                         .to(step2())
                                         .on("FAILED")
                                         .stop()
                                     .from(step1())
                                         .on("*")
                                         .to(step3())
                                         .next(step4())
                                     .from(step2())
                                         .on("*")
                                         .to(step5())
                                     .end()
                                     .build();
    }

결과

FAILED 되도록 코드 변경

// step1()
contribution.setExitStatus(ExitStatus.FAILED);

• 결과

[0JUUU]

사용자 정의 ExitStatus

• ExitStatus에 존재하지 않는 exitCode를 새롭게 정의해서 설정
• StepExecutionListener의 afterStep() 메서드 : Custom exitCode 생성 후 새로운 ExitStatus 반환
• Step 실행 후 완료 시점에서 현재 exitCode를 사용자 정의 exitCode로 수정 가능
• 리스너를 붙여 조작하는 방식보다 JobExecutionDecider를 사용하는게 더 좋다❗️

    @Bean
    public Job batchJob() {
        return jobBuilderFactory.get("batchJob")
            .start(step1())
                .on("FAILED")
                .to(step2())
                .on("PASS")
                .stop()
            .end()
            .build();
    }

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

• 왜 Job 이 FAILED?
  • Job Execution의 값은 마지막 Step 상태값
  • 현재 on("PASS").stop()만 함. 그니까 지금은 "PASS"상태가 아니라 stop을 돌지 않음
  • transition을 통해 값 할당 ➡️ 값에 만족하지 못할 경우에는 Step을 실패했다고 생각함
  • 조건에 해당하지 않는 경우에 대한 처리를 해줘야함

    .from(step2())
    .on("*")
    .stop()

    BATCH_JOB_EXECUTION

    

사용자 정의 Status

public class PassCheckingListener implements StepExecutionListener {

    @Override
    public void beforeStep(StepExecution stepExecution) {
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        String exitCode = stepExecution.getExitStatus().getExitCode();
        if (!exitCode.equals(ExitStatus.FAILED.getExitCode())) {
            return new ExitStatus("PASS");
        }
        return null;
    }
}

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

[0JUUU]

JobExecutionDecider

• ExitStatus를 조작하거나 StepExecutionListener를 등록할 필요 없이 Transition 처리를 위한 전용 클래스
• Step과 Transition 역할을 명확하게 분리 ⭕️
• 실행 여부를 결정하는 로직을 JobExecutionDecider를 구현하여 따로 설정할 수 있다.
• 우리가 커스텀한 상태값을 확인하고 처리할 수 있다.

🤔 왜 내가 정의한 ExitStatus & Listener의 조합보다 JobExecutionDecider 방식이 더 좋을까?

• Step : 비즈니스 로직에 집중
  • 하지만, ExitStatus & Listener의 조합은 Step 안에 Flow를 위한 ExitStatus 조작 코드가 들어가게 되어 그다지 좋은 방법이 아니다. (강한 결합, 가독성이 떨어짐 )
  • 그래서 on, to 이런 API가 아닌 내부 구현으로 복잡성을 해결할 수 있음

public class CustomDecider implements JobExecutionDecider {

    private int count = 0;

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        count++;
        System.out.println(">> count " + count);
        if(count % 2 == 0) {
            return new FlowExecutionStatus("EVEN");
        }
        return new FlowExecutionStatus("ODD");
    }
}

• 이런 식으로 짝수이면 EVEN 값을, 홀수라면 ODD 값을 갖도록 따로 설정할 수 있다.

private int count = 1;

• 여기서 설정한 pattern 값과 Status값이 일치하는지 확인 (SimpleFlow 내부에서)

[0JUUU]

FlowJob 아키텍처

[0JUUU]

SimpleFlow - 개념 및 API 소개

• 스프링배치에서 제공하는 Flow의 구현체 ➡️ 각 요소(Step, Flow, JobExecutionDecider) 들을 담고 있는 State를 실행시키는 도메인 객체
• FlowBuilder를 사용하여 생성, Transition과 조합하여 여러 개의 Flow 및 중첩 Flow를 만들어 Job을 구성

[0JUUU]

SimpleFlow 예제

@RequiredArgsConstructor
@Configuration
public class SimpleFlowConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return jobBuilderFactory.get("batchJob")
            .start(flow1())
                .on("COMPLETED")
                .to(flow2())
            .from(flow1())
                .on("FAILED")
                .to(flow3())
            .end()
            .build();
    }

    @Bean
    public Flow flow1() {
        FlowBuilder<Flow> flowBuilder = new FlowBuilder<>("flow1");
        flowBuilder.start(step1())
            .next(step2())
            .end();
        return flowBuilder.build();
    }

    @Bean
    public Flow flow2() {
        FlowBuilder<Flow> flowBuilder = new FlowBuilder<>("flow2");
        flowBuilder.start(flow3())
                   .next(step5())
                   .next(step6())
                   .end();
        return flowBuilder.build();
    }

    @Bean
    public Flow flow3() {
        FlowBuilder<Flow> flowBuilder = new FlowBuilder<>("flow3");
        flowBuilder.start(step3())
                   .next(step4())
                   .end();
        return flowBuilder.build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step1 has executed");
//                    throw new RuntimeException("step1 was failed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }

    @Bean
    public Step step2() {
        return stepBuilderFactory.get("step2")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step2 has executed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }

    @Bean
    public Step step3() {
        return stepBuilderFactory.get("step3")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step3 has executed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }

    @Bean
    public Step step4() {
        return stepBuilderFactory.get("step4")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step4 has executed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }

    @Bean
    public Step step5() {
        return stepBuilderFactory.get("step5")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step5 has executed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }

    @Bean
    public Step step6() {
        return stepBuilderFactory.get("step6")
                                 .tasklet((contribution, chunkContext) -> {
                                     System.out.println(">> step6 has executed");
                                     return RepeatStatus.FINISHED;
                                 }).build();
    }
}

• Step1 : Completed

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

일부러 예외 발생 (step1에서 실패 -> Flow 3로 감)

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

[0JUUU]

SimpleFlow 아키텍처

• 각각의 State : Job이 구성되는 시점에 타입에 맞게 FlowBuilder에 의해 생성
• 이런 State들이 on에서 패턴을 만나면, 다음에 실행될 State 등에 대한 정보가 StateTransition 객체에 담김
• SimpleFlow : 각각의 StateTransition을 List로 갖고 이를 통해 나머지 필드값 초기화

• 처음 실행될 StartState 지정하고 실행
• StartState 완료 ➡️ 다음 State 실행을 위해 resume() 메서드 호출
• resume() : 다음 State를 선택하기 위해 nextState() 호출
• nextState() : StateMap을 통해 다음 State 선택
• 결과적으로 resume() 메서드로 돌아와 다음 State 실행

[0JUUU]

FlowStep

• Step 내에서 Flow를 할당하여 실행시키는 도메인 객체 (c.f. JobStep)
• flowStep의 BatchStatus와 ExitStatus : Flow의 최종 상태값에 따라 결정 ➡️ FlowStep은 결국 Step이므로 결국 얘는 Job에 영향

@RequiredArgsConstructor
@Configuration
public class FlowStepConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return jobBuilderFactory.get("batchJob")
                                .start(flowStep())
                                .next(step2())
                                .build();
    }

    private Step flowStep() {
        return stepBuilderFactory.get("flowStep")
            .flow(flow())
            .build();
    }

    private Flow flow() {
        FlowBuilder<Flow> flowBuilder = new FlowBuilder<>("flow");
        flowBuilder.start(step1())
            .end();
        return flowBuilder.build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                 .tasklet(new Tasklet() {
                                     @Override
                                     public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                                         System.out.println("step1 was executed");
                                         throw new RuntimeException("step1 was failed");
//                        return RepeatStatus.FINISHED;
                                     }
                                 })
                                 .build();
    }

    @Bean
    public Step step2() {
        return stepBuilderFactory.get("step2")
                                 .tasklet(new Tasklet() {
                                     @Override
                                     public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                                         System.out.println("step2 was executed");
                                         return RepeatStatus.FINISHED;
                                     }
                                 })
                                 .build();
    }
}

BATCH_JOB_EXECUTION

BATCH_STEP_EXECUTION

• step이 실패하게 되면 Job도 실패한다. (상태값에 대한 연관관계)

[0JUUU]

@JobScope / @StepScope - 기본개념 및 설정

Scope

• 스프링 컨테이너에서 빈이 관리되는 범위 (기본 singleton으로 생성)

Spring Batch Scope

@JobScpe, @StepScope

• Job과 Step의 빈 생성과 실행에 관여하는 scope
• 프록시 모드를 기본값으로 하는 스코프 - @Scope(value = "job", proxyMode = ScopedProxyMode.TARGET_CLASS)
• 해당 스코프 선언 ➡️ 빈 생성이 어플리케이션 구동시점이 아닌 빈의 실행 시점에 이루어짐
  • @Values 주입 ➡️ 빈의 실행 시점에 값을 참조 (일종의 Lazy Binding)
  • @Value("#{jobParameters[파라미터명]}"), @Value("#{jobExecutionContext[파라미터명]}"), @Value("#{stepExecutionContext[파라미터명]}")
  • @Values 사용 시, @JobScope, @StepScope를 정의하지 않으면 오류 발생 ➡️ 반드시 선언 필수 ❗️

    @JobScope

• Step 선언문에 정의
• @Value : jobParameter, jobExecutionContext만 사용 가능

  @StepScope

• Tasklet 선언문에 정의
• @Value : jobParameter, jobExecutionContext, stepExecutionContext 사용 가능

CustomJobListener

public class CustomJobListener implements JobExecutionListener {

    @Override
    public void beforeJob(JobExecution jobExecution) {
        jobExecution.getExecutionContext().putString("name", "user1");
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
    }
}

CustomStepListener

public class CustomStepExecutionListener implements StepExecutionListener {

    @Override
    public void beforeStep(StepExecution stepExecution) {
        stepExecution.getExecutionContext().putString("name2", "user2");
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }
}

Configuration

@RequiredArgsConstructor
@Configuration
public class JobScope_StepScope_Configuration {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return jobBuilderFactory.get("batchJob")
                                .start(step1(null))
                                .next(step2())
                                .listener(new CustomJobListener())
                                .build();
    }

    @Bean
    @JobScope
    public Step step1(@Value("#{jobParameters['message']}") String message) {
        System.out.println("jobParameters['message'] : " + message);
        return stepBuilderFactory.get("step1")
                                 .tasklet(tasklet1(null))
                                 .build();
    }

    @Bean
    public Step step2() {
        return stepBuilderFactory.get("step2")
                                 .tasklet(tasklet2(null))
                                 .listener(new CustomStepExecutionListener())
                                 .build();
    }

    @Bean
    @StepScope
    public Tasklet tasklet1(@Value("#{jobExecutionContext['name']}") String name) {
        System.out.println(">> jobExecutionContext['name'] = " + name);
        return (stepContribution, chunkContext) -> {
            System.out.println(">> tasklet1 has executed");
            return RepeatStatus.FINISHED;
        };
    }

    @Bean
    @StepScope
    public Tasklet tasklet2(@Value("#{stepExecutionContext['name2']}") String name2) {
        System.out.println(">> stepExecutionContext['name2'] = " + name2);
        return (stepContribution, chunkContext) -> {
            System.out.println(">> tasklet2 has executed");
            return RepeatStatus.FINISHED;
        };
    }
}

@JobScope를 생략한다면? ➡️ 에러 발생

[0JUUU]

@JobScope / @StepScope 아키텍처

Proxy 객체 생성

• @JobScope, @StepScope 이 붙은 빈 선언 ➡️ 내부적으로 빈의 Proxy 객체 생성
• Job 실행 시 Proxy 객체가 실제 빈을 호출해서 해당 메서드를 실행시키는 구조 (AOP 개념 적용)

🤔 왜 Proxy 객체를 사용하지?

• 표현식 언어를 통해 유연하고 편리하게 주입받아 파라미터로 사용할 수 있음
  • StepConfiguration에서 일일이 원하는 값을 꺼내지 않고도 사용 가능
• Step 빈 생성이 구동시점이 아닌 런타임 시점에서 생성되어 객체의 지연 로딩 ⭕️
  • 표현식으로 적은 값 : 컴파일 시점에 존재 ❌, 런타임 시점에 채워지면서 존재 ⭕️
  • 만약 빈이 애플리케이션 로딩 시점에 만들어진다면 DI를 해야하는데 해당 값들이 현재 존재하지 않기 때문에 찾을 수 ❌
  • 하지만 런타임 시점에 빈을 만들게 되면 값을 다 받아놓고(표현식에 명시) 빈을 만들어 주입 가능
• 병렬 처리 시 각 스레드마다 Step 객체가 생성되어 할당 ➡️ Tasklet에 멤버 변수가 존재해도 동시성에 문제 ❌

JobScope, StepScope

• Proxy 객체의 실제 대상이 되는 Bean 등록, 해제하는 역할
• 실제 빈을 저장하고 있는 JobContext, StepContext를 가짐

JobContext, StepContext

• 스프링 컨테이너에서 생성된 빈을 저장하는 컨텍스트 역할
• Job의 실행 시점에서 프록시 객체가 실제 빈을 참조할 때 사용