TIL

02. Controlling Step Flow

• job 내에서 step을 그룹화할 수 있는 기능을 사용하면 job이 한 step에서 다른 step으로 흐르는 방식을 제어할 수 있어야 한다.
• 한 Step의 실패가 반드시 Job의 실패를 의미하는 것은 아니다.
• Step 그룹이 어떻게 구성되는지에 따라 특정 Step은 전혀 처리되지 않을 수도 있다.

Sequential Flow

• 가장 간단한 시나리오는 순차적으로 처리되는 흐름이다.

• next()로 Step을 설정하면 된다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    				.start(stepA())
    				.next(stepB())
    				.next(stepC())
    				.build();
    }
  

  • Step A, B, C 순으로 실행된다.
  • 앞 단계의 Step이 실패하면 전체 Job은 실패하고 다음 Step은 실행되지 않는다.

Conditional Flow

• 순차적인 실행 흐름에선 Step 성공 후 다음 Step으로 이어지거나 실패해서 다음 Step이 실행되지 않는 두 시나리오만 있었다.
• 하지만 한 Step 실패 후 다른 Step을 트리거해야 하는 시나리오가 필요할 수도 있다.

• Java API를 사용해서 실패 시 수행할 Step을 지정할 수 있다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    				.start(stepA())
    				.on("*").to(stepB())
    				.from(stepA()).on("FAILED").to(stepC())
    				.end()
    				.build();
    }
  

  • on() 메서드는 간단한 패턴 매칭 체계를 사용하여 Step 실행 결과인 ExitStatus를 검사한다.
    • *: 0개 이상의 문자와 일치
    • ?: 정확히 한 문자와 일치
    • c*t는 cat, count와 매칭되지만 c?t는 cat이랑만 매칭된다.
  • Step 실행 결과인 ExitStatus가 요소에 매칭되지 않으면 예외가 발생하고 Job이 실패한다.
  • 위 예제에선 Step A가 실패해도 Step C가 실행된다.

Batch Status Versus Exit Status

Conditional Flow를 구성할 때 BatchStatus와 ExitStatus 차이점을 이해하는 것이 중요하다.

• BatchStatus는 JobExecution과 StepExecution의 속성인 Enum이며 Job 또는 Step 상태를 기록하는 데 사용된다.
  • COMPLETED, STARTING, STARTED, STOPPING, STOPPED, FAILED, ABANDONED, UNKNOWN

• on을 사용한 Java Configuration

    ...
    .from(stepA()).on("FAILED").to(stepB())
    ...
  

  • 위 코드는 BatchStatus가 아닌 ExitStatus를 참조한다.
  • Step A가 실패하는 경우 Step B로 이동이라는 의미를 나타낸다.

• 아래 step1엔 세 가지 가능성이 존재한다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    			.start(step1()).on("FAILED").end()
    			.from(step1()).on("COMPLETED WITH SKIPS").to(errorPrint1())
    			.from(step1()).on("*").to(step2())
    			.end()
    			.build();
    }
  

  • Step이 실패하여 Job이 실패하는 경우
  • Step이 성공하는 경우
  • Step이 성공했지만 exit code가 COMPLETED WITH SKIPS이기 때문에 오류를 처리하기 위해 다른 Step을 실행하게 된다.

• 적절한 exit code를 반환하도록 아래 코드를 작성할 수 있다.

    public class SkipCheckingListener extends StepExecutionListenerSupport {
        public ExitStatus afterStep(StepExecution stepExecution) {
            String exitCode = stepExecution.getExitStatus().getExitCode();
            if (!exitCode.equals(ExitStatus.FAILED.getExitCode()) &&
                  stepExecution.getSkipCount() > 0) {
                return new ExitStatus("COMPLETED WITH SKIPS");
            }
            else {
                return null;
            }
        }
    }
  

  • Step이 성공했는지 확인한 후 다음 Step 실행의 스킵 횟수가 0보다 큰지 확인하는 StepExecutionListener 코드다.
  • 두 조건 모두 충족하면 COMPLETED WITH SKIPS를 반환한다.

Configuring for Stop

• BatchStatus와 ExitStatus는 실행되는 코드인 Step에 의해 결정되는 반면 Job의 상태는 설정에 따라 결정된다.
• 지금까지 설명한 모든 job 설정에는 전환이 없는 최종 Step이 하나 이상 포함되어 있다.

• 아래 예제는 Job이 Step 실행 후 끝난다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    				.start(step1())
    				.build();
    }
  

  • Step에 대해 전환이 정의되지 않은 경우 Job 상태는 다음과 같이 정의된다.
  • Step이 ExistStatus FAILED로 끝나는 경우 BatchStatus도 FAILED가 된다.
  • 그렇지 않으면 둘 다 COMPLETED가 된다.
• 위 코드의 경우 간단한 순차적 작업에는 충분할 수 있지만 커스텀한 job-stopping 시나리오가 필요할 수도 있다.
• 스프링 배치는 next()를 포함하여 Job을 멈추는 세 가지 전환 요소를 제공한다.
  • 이 요소들은 특정 BatchStatus와 함께 Job을 죵료시킨다.
  • 전환 요소는 Job의 최종 상태에만 영향을 주고 어떤 Step의 BatchStatus 또는 ExitStatus에 영향을 미치지 않는다.
  • 예를 들어 Job의 모든 Step이 FAILED라도 Job 상태는 COMPLETED일 수 있다.

Ending at a Step

• step에 end를 설정하면 BatchStatus가 COMPLETED가 되도록 Job을 중지한다.
  • 완료된 JOB은 재시작될 수 없다. (JobInstanceAlreadyCompleteException을 던짐)
• end() 메서드를 사용해서 step을 끝낼 수 있고 선택적 exitStatus 매개 변수를 허용한다.
  • exiitStatus가 제공되지 않으면 기본적으로 ExitStatus가 BatchStatus와 일치되어 COMPLETED로 설정된다.

@Bean
public Job job(JobRepository jobRepository) {
	return new JobBuilder("job", jobRepository)
				.start(step1())
				.next(step2())
				.on("FAILED").end()
				.from(step2()).on("*").to(step3())
				.end()
				.build();
}

Failing a Step

• 특정 시점에 실패하도록 설정하면 Job의 BatchStatus가 FAILED가 된다.
  • 종료와 달리 실패한 Job의 재시작을 막지 않는다.

@Bean
public Job job(JobRepository jobRepository) {
	return new JobBuilder("job", jobRepository)
			.start(step1())
			.next(step2()).on("FAILED").fail()
			.from(step2()).on("*").to(step3())
			.end()
			.build();
}

Stopping a Job at a Given Step

• 특정 step에서 작업을 중지하도록 구성하면 BatchStatus가 STOPPED로 중지되도록 할 수 있다.
  • Job을 중지하면 처리가 일시적으로 중단되어 작업자가 작업을 다시 시작하기 전에 몇 가지 조치를 취할 수 있다.

• stopAndRestart 메서드로 작업이 다시 시작될 때 실행이 재개될 step을 지정하는 restart 속성을 필요로 한다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    			.start(step1()).on("COMPLETED").stopAndRestart(step2())
    			.end()
    			.build();
    }
  

  • step1()이 COMPLETED로 끝나면 작업이 중지된다.
  • 다시 시작되면 step2()에서 실행이 시작된다.

Programmatic Flow Decisions

• 어떤 상황에선 다음 step을 결정하기 위해 ExitStatus보다 더 많은 정보가 필요할 수 있다.
• 다음 예제에서 JobExecutionDecider를 사용해서 결정을 지원할 수 있다.

public class MyDecider implements JobExecutionDecider {
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        String status;
        if (someCondition()) {
            status = "FAILED";
        }
        else {
            status = "COMPLETED";
        }
        return new FlowExecutionStatus(status);
    }
}

@Bean
public Job job(JobRepository jobRepository) {
	return new JobBuilder("job", jobRepository)
			.start(step1())
			.next(decider()).on("FAILED").to(step2())
			.from(decider()).on("COMPLETED").to(step3())
			.end()
			.build();
}

Split Flows

• 지금까지는 선형적인 방식으로 한 step씩 실행하는 Job이었지만 병렬 흐름으로도 구성할 수 있다.
• split()은 하나 이상의 flow 요소를 포함하며 전체 개별 흐름을 정의할 수 있다.
• split()은 next 속성이나 next, end, 그리고 fail 속성과 같이 전환 요소도 포함될 수 있다.
• 아래 예제에선 flow1과 flow2가 병렬 실행되고 마지막으로 step4가 실행된다.

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

@Bean
public Flow flow2() {
	return new FlowBuilder<SimpleFlow>("flow2")
			.start(step3())
			.build();
}

@Bean
public Job job(Flow flow1, Flow flow2) {
	return this.jobBuilderFactory.get("job")
				.start(flow1)
				.split(new SimpleAsyncTaskExecutor())
				.add(flow2)
				.next(step4())
				.end()
				.build();
}

Externalizing Flow Definitions and Dependencies Between Jobs

• 작업 흐름 중 일부를 별도의 빈 정의로 외부화해서 재사용할 수 있다.

• 첫 번째 방법으로 flow를 다른 곳에 정의된 flow에 대한 참조로 선언하는 방법이 있다.

    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    				.start(flow1())
    				.next(step3())
    				.end()
    				.build();
    }
      
    @Bean
    public Flow flow1() {
    	return new FlowBuilder<SimpleFlow>("flow1")
    			.start(step1())
    			.next(step2())
    			.build();
    }
  

  • 마치 인라인으로 선언된 것처럼 외부 flow 단계를 job에 삽입하는 효과를 얻을 수 있다.
  • 많은 작업이 동일한 탬플릿 flow를 참조하고 서로 다른 논리적 flow를 구성할 수도 있다.
  • 개별 flow의 통합 테스트를 분리하는 좋은 방법이기도 하다.
• 두 번째 방법으로 JobStep을 사용하는 것이 있다.

  • FlowStep과 유사하지만 실제로는 지정된 flow 단계에 대해 별도의 job execution을 생성하고 실행한다.

    @Bean
    public Job jobStepJob(JobRepository jobRepository) {
    	return new JobBuilder("jobStepJob", jobRepository)
    				.start(jobStepJobStep1(null))
    				.build();
    }
        
    @Bean
    public Step jobStepJobStep1(JobLauncher jobLauncher, JobRepository jobRepository) {
    	return new StepBuilder("jobStepJobStep1", jobRepository)
    				.job(job())
    				.launcher(jobLauncher)
    				.parametersExtractor(jobParametersExtractor())
    				.build();
    }
        
    @Bean
    public Job job(JobRepository jobRepository) {
    	return new JobBuilder("job", jobRepository)
    				.start(step1())
    				.build();
    }
        
    @Bean
    public DefaultJobParametersExtractor jobParametersExtractor() {
    	DefaultJobParametersExtractor extractor = new DefaultJobParametersExtractor();
        
    	extractor.setKeys(new String[]{"input.file"});
        
    	return extractor;
    }
    

  • jobParametersExtractor는 실행되는 Job에 대한 ExecutionContext가 JobParameters로 변환되는 방법을 결정하는 전략이다.
  • JobStep은 큰 시스템을 더 작은 모듈로 나누고 작업 흐름을 제어할 수 있는 좋은 방법이다.

Late Binding of Job and Step Attributes

@Bean
public FlatFileItemReader flatFileItemReader() {
	FlatFileItemReader<Foo> reader = new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource("file://outputs/file.txt"))
			...
}

• Resource는 지정된 파일 시스템 위치에서 파일을 로드한다.
  • 절대 경로는 //로 시작해야 한다.
• 대부분의 Spring 애플리케이션에선 이러한 리소스 이름을 컴파일 시점에 알 수 있기에 위 방법으로 충분하다.
• 그러나 배치 시나리오에선 파일 이름은 런타임에 job parameters로 결정해야할 수도 있다.

• -D 매개 변수를 통새 시스템 property를 읽으면 해결할 수 있다.

    @Bean
    public FlatFileItemReader flatFileItemReader(@Value("${input.file.name}") String name) {
    	return new FlatFileItemReaderBuilder<Foo>()
    			.name("flatFileItemReader")
    			.resource(new FileSystemResource(name))
    			...
    }
  

  • Dinput.file.name="file://outputs/file.txt”

• 시스템 property를 통하지 않고 JobParameters를 바인딩할 수 있다.

    @StepScope
    @Bean
    public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters['input.file.name']}") String name) {
    	return new FlatFileItemReaderBuilder<Foo>()
    			.name("flatFileItemReader")
    			.resource(new FileSystemResource(name))
    			...
    }
  

• 동일한 방식으로 JobExecution 및 StepExecution 수준의 ExecutionContext에 모두 접근할 수 있다.

    @StepScope
    @Bean
    public FlatFileItemReader flatFileItemReader(@Value("#{jobExecutionContext['input.file.name']}") String name) {
    	return new FlatFileItemReaderBuilder<Foo>()
    			.name("flatFileItemReader")
    			.resource(new FileSystemResource(name))
    			...
    }
  

    @StepScope
    @Bean
    public FlatFileItemReader flatFileItemReader(@Value("#{stepExecutionContext['input.file.name']}") String name) {
    	return new FlatFileItemReaderBuilder<Foo>()
    			.name("flatFileItemReader")
    			.resource(new FileSystemResource(name))
    			...
    }
  

Step Scope

• 앞서 표시된 모든 지연 바인딩 예제에는 빈 정의에 선언된 스코프가 있었다.

@StepScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters[input.file.name]}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

• Step이 시작될 때까지 빈을 인스턴스화하지 않기 때문에 지연 바인딩을 사용하려면 @StepScope를 사용해야 한다.
• 기본적으로 스프링 컨테이너의 일부가 아니기에 다음의 방법이 필요하다.
  • batch namespace를 사용
  • @StepScope로 빈 정의
  • @EnableBatchProcessing을 사용하여 범위를 명시적으로 추가

JobScope

• Spring Batch 3.0에 도입된 @JobScope는 @StepScope와 유사하지만 Job 컨텍스트에 대한 범위이므로 실행 중인 작업 당 해당 빈의 인스턴스가 하나만 존재한다.
  • #{..} 표현을 사용하여 JobContext에서 접근할 수 있는 참조의 지연 바인딩도 지원된다.
  • job이나 job 실행 컨텍스트와 job 매개변수에서 빈 속성을 가져올 수 있다.

@JobScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters[input]}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

@JobScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{jobExecutionContext['input.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

• 스프링 컨테이너 일부가 아니기 때문에 다음의 방법이 필요하다.
  • batch namespace 사용
  • @JobScope에 대한 빈 정의 명시적 선언
  • @EnableBatchProcessing을 사용하여 범위를 명시적으로 추가

Scoping ItemStream components

• ItemStream 빈을 job 또는 step scope에 정의하는 경우 반환 유형은 최소한 ItemStream이어야 한다.
• 이는 스프링 배치가 이 인터페이스를 구현하는 프록시를 올바르게 생성하여 예상대로 open, update, close 메서드르 호출하여 계약을 준수하기 위해서다.

@Bean
@StepScope
public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters['input.file.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.resource(new FileSystemResource(name))
			// set other properties of the item reader
			.build();
}