Spring Batch Step Configuration

개요

Chunk-oriented Processing

Cofiguring a Step

@Bean
public Step sampleStep(JobRepository jobRepository, 
		PlatformTransactionManager transactionManager) { 
	return new StepBuilder("sampleStep", jobRepository)
				.<String, String>chunk(10, transactionManager) 
				.reader(itemReader())
				.writer(itemWriter())
				.build();
}

Configuring a Step for Restart

Setting a Start Limit

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.startLimit(1)
				.build();
}

Restarting a Completed Step

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.allowStartIfComplete(true)
				.build();
}

Configuring Skip Logic

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(flatFileItemReader())
				.writer(itemWriter())
				.faultTolerant()
				.skipLimit(10)
				.skip(FlatFileParseException.class)
				.build();
}

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(flatFileItemReader())
				.writer(itemWriter())
				.faultTolerant()
				.skipLimit(10)
				.skip(Exception.class)
				.noSkip(FileNotFoundException.class)
				.build();
}

Configuring Retry logic

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(2, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.faultTolerant()
				.retryLimit(3)
				.retry(DeadlockLoserDataAccessException.class)
				.build();
}

Controlling Rollback

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(2, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.faultTolerant()
				.noRollback(ValidationException.class)
				.build();
}

Transactional Readers

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(2, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.readerIsTransactionalQueue()
				.build();
}

Transaction Attributes

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
	attribute.setPropagationBehavior(Propagation.REQUIRED.value());
	attribute.setIsolationLevel(Isolation.DEFAULT.value());
	attribute.setTimeout(30);

	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(2, transactionManager)
				.reader(itemReader())
				.writer(itemWriter())
				.transactionAttribute(attribute)
				.build();
}

Registering ItemStream with a  Step

ItemStream

• 자동 등록: ItemReader, ItemWriter, ItemProcessor가 직접 ItemStream을 구현한 경우, Spring Batch는 이를 자동으로 인식하고 등록한다.

• 수동 등록 필요: 그러나 CompositeItemWriter와 같이 내부에 여러 ItemWriter를 위임(delegation)하는 구조에서는, 위임된 구성 요소들이 ItemStream을 구현하더라도 자동으로 등록되지 않는다. 이 경우, 각 구성 요소를 명시적으로 Step에 등록해야 한다.

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("step1", jobRepository)
        .<String, String>chunk(2, transactionManager)
        .reader(itemReader())
        .writer(compositeItemWriter())
        .stream(fileItemWriter1()) // 수동 등록
        .stream(fileItemWriter2()) // 수동 등록
        .build();
}

@Bean
public CompositeItemWriter<String> compositeItemWriter() {
    CompositeItemWriter<String> writer = new CompositeItemWriter<>();
    writer.setDelegates(Arrays.asList(fileItemWriter1(), fileItemWriter2()));
    return writer;
}

Intercepting Step Execution

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(reader())
				.writer(writer())
				.listener(chunkListener())
				.build();
}

StepExecutionListener

public interface StepExecutionListener extends StepListener {

    void beforeStep(StepExecution stepExecution);

    ExitStatus afterStep(StepExecution stepExecution);

}

ChunkListner

public interface ChunkListener extends StepListener {

    void beforeChunk(ChunkContext context);
    void afterChunk(ChunkContext context);
    void afterChunkError(ChunkContext context);

}

ItemReadListener<T>

public interface ItemReadListener<T> extends StepListener {

    void beforeRead();
    void afterRead(T item);
    void onReadError(Exception ex);

}

ItemProcessListener<T, S>

public interface ItemProcessListener<T, S> extends StepListener {

    void beforeProcess(T item);
    void afterProcess(T item, S result);
    void onProcessError(T item, Exception e);

}

ItemWriteListener<S>

public interface ItemWriteListener<S> extends StepListener {

    void beforeWrite(List<? extends S> items);
    void afterWrite(List<? extends S> items);
    void onWriteError(Exception exception, List<? extends S> items);

}

SkipListener<T, S>

public interface SkipListener<T,S> extends StepListener {

    void onSkipInRead(Throwable t);
    void onSkipInProcess(T item, Throwable t);
    void onSkipInWrite(S item, Throwable t);

}

TaskletStep

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new StepBuilder("step1", jobRepository)
    			.tasklet(myTasklet(), transactionManager)
    			.build();
}

Tasklet

public class FileDeletingTasklet implements Tasklet, InitializingBean {

    private Resource directory;

    public RepeatStatus execute(StepContribution contribution,
                                ChunkContext chunkContext) throws Exception {
        File dir = directory.getFile();
        Assert.state(dir.isDirectory(), "The resource must be a directory");

        File[] files = dir.listFiles();
        for (int i = 0; i < files.length; i++) {
            boolean deleted = files[i].delete();
            if (!deleted) {
                throw new UnexpectedJobExecutionException("Could not delete file " +
                                                          files[i].getPath());
            }
        }
        return RepeatStatus.FINISHED;
    }

    public void setDirectoryResource(Resource directory) {
        this.directory = directory;
    }

    public void afterPropertiesSet() throws Exception {
        Assert.state(directory != null, "Directory must be set");
    }
}

• RepeatStatus.FINISHED

• RepeatStatus.CONTINUABLE

TaskletAdapter

@Bean
public Tasklet taskletAdapter() {
    MethodInvokingTaskletAdapter adapter = new MethodInvokingTaskletAdapter();
    adapter.setTargetObject(myService);
    adapter.setTargetMethod("performTask");
    return adapter;
}

Controlling Step Flow

Sequential Flow

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

Conditional Flow

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

• * : 0개 이상의 문자와 일치

• ? : 정확히 한 문자와 일치

• 하나의 Step에 여러 개의 분기를 지정 할 수 있으며, 그 수에는 제한이 없다.

• 하지만 Step 실행 결과로 반환된 ExitStatus가 지정된 어떤 조건에도 일치하지 않는 경우, SpringBatch는 예외를 던지고 전체 Job이 실패하게 된다.

• 분기 조건은 프레임워크에서 가장 구체적인 것부터, 덜 구체적인 순으로 정렬한다. 이 말은 즉슨, 분기 순서가 코드 상 먼저인지 여부와 상관없이, 더 정확히 일치하는 조건에 대해 결정된다.

• 분기를 설정할 때는, BatchStatus 와 ExitStatus 의 차이점을 정확하게 이해하는 것이 중요하다. 이는 바로 뒤에서 설명하겠다.

BatchStatus vs ExitStatus

Configuring for Stop

• end : Job을 성공적으로 종료한다. BatchStatus와 ExitStatus는 기본적으로 COMPLETED로 설정된다.

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

• fail : Job을 실패로 종료한다. BatchStatus는 FAILED로 설정된다.

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

• stop : Job을 중단하고, 이후 재시작 시 특정 Step부터 실행하도록 설정한다. BatchStatus는 STOPPED로 설정된다.

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

Programmatic Flow Decisions

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, MyDecider decider, Step step1, Step step2, Step step3) {
	return new JobBuilder("job", jobRepository)
			.start(step1)
			.next(decider).on("FAILED").to(step2)
			.from(decider).on("COMPLETED").to(step3)
			.end()
			.build();
}

Split Flows

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

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

@Bean
public Job job(JobRepository jobRepository, Flow flow1, Flow flow2, Step step4) {
	return new JobBuilder("job", jobRepository)
				.start(flow1)
				.split(new SimpleAsyncTaskExecutor())
				.add(flow2)
				.next(step4)
				.end()
				.build();
}

1. flow1 실행
→ step1 → step2 (순차 실행)

2. 동시에 flow2도 실행
→ step3
☑ 병렬 처리는 .split()과 SimpleAsyncTaskExecutor로 구현됨
→ flow1과 flow2는 병렬로 동시에 실행됨

3. 두 Flow가 모두 완료된 후, step4 실행

마무리