Spring Batch (4) Job, JobInstance, JobParameters, JobExecution

📎 Job, JobInstance, JobParameters, JobExecution

지난 시간에는 Spring Batch의 Meta Data Schema에 대해 살펴보았습니다.

이번 시간에는 Job의 여러 도메인 객체인 JobInstance, JobParameters, JobExecution 객체에 대해 살펴보겠습니다.

(예제 코드는 GitHub에서 확인할 수 있습니다. 😃)

 

 

📌 해당 포스팅의 목표

• Spring Batch Job
• Spring Batch JobInstance
• Spring Batch JobParameters
• Spring Batch JobExecution

 

 

📌 Job

기본 개념

• 배치 계층 구조에서 최상위의 개념으로 전체 배치 프로세스를 캡슐화한 엔티티
• Job Configuration을 통해 생성되는 객체 단위로 배치 작업을 어떻게 구성하고 실행할지 명세해 놓은 객체 
• 여러 Step들을 포함하는 컨테이너로 반드시 한 개 이상의 Step으로 구성해야 함

 

기본 구현체

 

AbstractJob

Job을 구현하고 있는 추상 클래스로 여러 필드 및 메서드들이 존재합니다.

• name: Job 이름
• restartable: Job의 재시작 여부(default = true)
• JobRepository: 메타 데이터 저장소
• JobExecutionListener: Job의 이벤트 리스너
• JobParametersIncrementer: JobParameters 증가기
• JobParametersValidator: JobParameters 검증기
• StepHandler: Step 실행 핸들러

 

SimpleJob

• 순차적으로 Step을 실행시키는 Job
• 모든 Job에서 유용하게 사용할 수 있는 표준 기능을 포함

 

FlowJob

• 특정 조건 및 흐름에 따라 Step을 구성하여 실행하는 Job
• Flow 객체를 실행시켜 작업을 진행

 

 

📌 JobInstance

기본 개념

• Job이 실행되는 시점에서 생성되는 Job의 논리적 작업 실행 개념
• Job이 실행되는 시점에 처리하는 내용은 각각 다르기 때문에 Job의 실행을 구분해야 함
  • 예를 들어, 하루에 한 번 매일 Job이 실행된다면, 실행되는 각각의 Job을 JobInstance라고 함
• 처음 시작하는 Job & JobParameters의 경우 새로운 JobInstance를 생성
• 이전과 동일한 Job & JobParameters으로 실행할 경우 이미 존재하는 JobInstance를 리턴
• Meta Data Schema의 BATCH_JOB_INSTANCE 테이블과 매핑

 

 

JobInstance 과정

(참고: 인프런 Spring Batch)

 

JobLauncher는 실제 Batch Job을 실행시키는 Launcher로 Job을 실행할 때(run) Job, JobParameters 두 개의 인자를 받아서 Spring Batch Job을 실행합니다.

 

위 사진처럼 Job = 일별 정산, JobParameters = '2021.01.01' 이라는 인자를 받아 Job을 실행한다고 했을 때, JobRepository라는 클래스(메타 데이터의 저장소 역할)가 Job & JobParameters를 Key로 해서 이전에 실행한 내역이 존재하는지 DB에서 조회합니다.

 

Job & JobParameters로 조회 후 JobInstance가 존재할경우, 기존 JobInstance를 리턴하고 존재하지 않을 경우 새로운 JobInstance를 생성합니다.

(JobParameters는 Spring Batch가 실행될 때 외부에서 받는 파라미터를 의미하는데요, 뒤에서 자세히 살펴보겠습니다.)

 

 

예제 코드

(코드 편의 상 import는 모두 제외했습니다.)

@RequiredArgsConstructor
@Configuration
public class JobInstanceConfiguration {
 
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
 
    @Bean
    public Job instanceBatchJob() {
        return this.jobBuilderFactory.get("Job")
                .start(instanceStep1())
                .next(instanceStep2())
                .build();
    }
 
    @Bean
    public Step instanceStep1() {
        return stepBuilderFactory.get("instanceStep1")
                .tasklet((contribution, chunkContext) -> {
                    JobInstance jobInstance = contribution.getStepExecution().getJobExecution().getJobInstance();
                    System.out.println("jobInstance.getId() : " + jobInstance.getId());
                    System.out.println("jobInstance.getInstanceId() : " + jobInstance.getInstanceId());
                    System.out.println("jobInstance.getJobName() : " + jobInstance.getJobName());
                    System.out.println("jobInstance.getJobVersion : " + jobInstance.getVersion());
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
 
    @Bean
    public Step instanceStep2() {
        return stepBuilderFactory.get("instanceStep2")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("step2 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

하나의 Job과 2개의 Step을 가지는 간단한 애플리케이션 코드를 위와 같이 작성해 주었습니다.

JobInstance는 StepContribution 객체로부터 가져와 사용할 수 있습니다.

 

 

@Component
public class JobRunner implements ApplicationRunner {

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job job;

    @Override
    public void run(ApplicationArguments args) throws Exception {

        JobParameters jobParameters = new JobParametersBuilder()
                .addString("name", "JuHyun")
                .toJobParameters();

        jobLauncher.run(job, jobParameters);
    }
}

JobParameters 확인을 위해 실제 JobLauncher를 통해 위에서 작성한 Batch Job을 실행시키는 코드를 작성합니다.

 

 

spring:
  batch:
    job:
      enabled: false
      
 ---
 spring.batch.job.enabled=false

그 후 application.yml 파일에 Spring Boot가 자동으로 실행하는 Job 설정을 false로 설정합니다.

 

위 코드를 작성한 후 애플리케이션을 실행해보면 Job이 정상적으로 실행이 되고 데이터베이스에 저장이 됩니다.

BATCH_JOB_INSTANCE

BATCH_JOB_EXECUTION_PARAMS

 

동일한 코드 및 JobParameters로 두 번째 실행하면 다음과 같이 파라미터를 변경하는 에러와 함께 배치 실행이 실패하게 됩니다.

에러 문구를 확인해 보면 JobInstance가 이미 존재하기 때문에 발생한다고 나와있고, Job을 다시 실행하고 싶은 경우 parameters를 변경하라고 나와있기 때문에 JobParameters를 변경한 후 다시 실행해 보면 정상적으로 성공합니다.

 

 

 

📌 JobParameters

기본 개념

• Spring Batch의 Job을 실행할 때 외부에서 받는 파라미터를 가진 도매인 객체
• 하나의 Job에 존재할 수 있는 여러개의 JobInstance를 구분하기 위한 용도
• JobInstance와 JobParameters는 1:1 관계
• Meta Data Schema의 BATCH_JOB_EXECUTION_PARAM 테이블과 매핑

 

생성 및 바인딩

• 애플리케이션 실행 시 주입 : java -jar ~.jar -requestDate=20210101
• 코드로 생성 : JobParameterBuilder
• SPEL 사용 : @Value(“#{jobParameter[requestDate]}”) - @JobScope, @StepScope 선언이 필수
  
  • @JobScope, @StepScope는 Job, Step이 실행되는 시점에 Bean이 생성되도록 Bean의 생성시점을 지연시키는 역할을 합니다. @Value 어노테이션의 경우 애플리케이션이 실행될 때 바인딩을 하기 때문에 @JobScope, @Stepcope 어노테이션이 존재하지 않으면 NullPointerException 예외가 발생합니다.

 

예제 코드 1

@RequiredArgsConstructor
@Configuration
public class JobParameterConfiguration {
 
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
 
    @Primary
    @Bean
    public Job BatchJob() {
        return this.jobBuilderFactory.get("JobParameters")
                .start(step1())
                .next(step2())
                .build();
    }
 
    @Bean
    public Step step1() {
        return stepBuilderFactory.get("JobParametersStep1")
                .tasklet((contribution, chunkContext) -> {
                    JobParameters jobParameters = contribution.getStepExecution().getJobParameters();
                    String name = jobParameters.getString("name");
                    Long seq = jobParameters.getLong("seq");
                    Date date = jobParameters.getDate("date");
 
                    System.out.println("===========================");
                    System.out.println("name: " + name);
                    System.out.println("seq: " + seq);
                    System.out.println("date: " + date);
                    System.out.println("===========================");
 
                    Map<String, Object> jobParameters2 = chunkContext.getStepContext().getJobParameters();
                    String name2 = (String) jobParameters2.get("name");
                    long seq2 = (long) jobParameters2.get("seq");
 
                    System.out.println("name: " + name2);
                    System.out.println("seq: " + seq);
 
                    System.out.println("step1 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
 
    @Bean
    public Step step2() {
        return stepBuilderFactory.get("JobParametersStep2")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("step1 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

@Component
public class JobParameterTest implements ApplicationRunner {
 
    @Autowired
    JobLauncher jobLauncher;
 
    @Autowired
    Job job;
 
    @Override
    public void run(ApplicationArguments args) throws Exception {
 
        JobParameters jobParameters = new JobParametersBuilder().addString("name", "JuHyuns")
                .addLong("seq", 1L)
                .addDate("date", new Date())
                .toJobParameters();
 
        jobLauncher.run(job, jobParameters);
    }
}

위 예제 코드에서 contribution을 이용하여 JobParameters를 가져오는 코드는 다음과 같습니다.

JobParameters jobParameters = contribution.getStepExecution().getJobParameters();

contribution은 StepContribution 클래스이고, StepContribution 클래스 내부에 StepExecution 필드를 가지고 있습니다. 또한 StepExecution 클래스 내부에는 JobExecution라는 필드가 존재하고, 최종적으로 JobExecution 클래스 내에 JobParameters 필드를 가지고 있기 때문에 JobParameters를 참조할 수 있습니다. 즉, 다음과 같이 참조하고 있습니다.

StepContribution -> StepExecution -> JobExecution (JobParameters)

 

또한 chunkContext를 통해서도 JobParameters를 가져올 수 있습니다.

Map<String, Object> jobParameters2 = chunkContext.getStepContext().getJobParameters();

contribution와는 다르게 타입이 JobParameters이 아닌 아닌 Map 형식으로 가져오게 됩니다.

따라서 첫 번째 방식인 StepContribution 객체를 통해 JobParameters를 가져오는 것이 보다 좋은 방법일 듯합니다.

 

 

예제 코드 2

@RequiredArgsConstructor
@Configuration
public class JobParameterConfiguration {
 
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
 
    @Primary
    @Bean
    public Job batchJob() {
        return this.jobBuilderFactory.get("JobParameters")
                .start(step1(null))
                .next(step2())
                .build();
    }
 
    @Bean
    @JobScope // Scope 추가(지연 초기화)
    public Step step1(@Value("#{jobParameters[requestDate]}") String requestDate) {
        return stepBuilderFactory.get("JobParametersStep1")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("=======================");
                    System.out.println("requestDate: " + requestDate);
                    System.out.println("=======================");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
 
    @Bean
    public Step step2() {
        return stepBuilderFactory.get("JobParametersStep2")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("step1 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}
 

기본적으로 예제 코드 1과 비슷하지만, SPEL 적용을 위해 step메서드에 상단에 @JobScope 어노테이션과 메서드 인자에 @Value 어노테이션을 통해 JobParameters의 정보를 추가해주었습니다.

 

추가적으로 애플리케이션 실행 시 인자를 전달하기 위해 Edit Configurations > Modify options > Program arguments를 통해 파라미터를 전달한 후 Spring Batch Job을 실행해보면 정상적으로 값을 받아올 수 있습니다.

참고로 JobLauncher가 아닌 애플리케이션 실행 시 Batch를 실행하려면 application.yml 파일에서 아래와 같이 설정을 하면 됩니다.

spring:
  batch:
    job:
      enabled: true
      names: JobParameters
  profiles:
    active: local

 

JobParameters를 외부에서 전달받음으로써 얻는 이점은 굉장한데요, 특정 값에 따라 분기 처리가 필요한 내부 로직 등의 아래 예시와 같은 경우 굉장히 유용하게 활용할 수 있습니다.

• 날짜별 분기 처리가 필요한 경우
• Spring Batch Job의 실행 주체가 Jenkins로 관리될 경우, Jenkins 빌드 시 동적 파라미터를 받아 처리해야 할 경우
• 인터페이스의 타입에 따라 동적으로 로직, 경로 등이 변경되어야 하는 경우

 

 

 

📌 JobExecution

기본 개념

• JobExecution은 Job에 대한 한 번의 시도를 의미하는 객체로 Job 실행 중 발생한 정보들을 저장하고 있는 객체
  • 시작 시각, 종료 시각, 상태, 종료 상태 등의 정보
• JobInstance와의 관계
  • JobExecution의 실행 상태 결과가 'COMPLETED' -> Job 재 실행 불가능
  • JobExecution의 실행 상태 결과가 'FAILED' -> Job 재 실행 가능
  • JobExecution의 실행 상태 결과가 'COMPLETED'가 될 때까지 하나의 JobInstance 내에서 여러 번 시도 가능
  • JobInstance와 JobExecution는 1 : N의 관계  
• BATCH_JOB_EXECUTION 테이블과 매핑

 

JobExecution 과정

(참고: 인프런 Spring Batch)

1. JobLauncher가 Batch Job을 실행(Job & JobParameters)
2. JobRepository가 Job & JobParameters로 DB 조회
3. JobInstance가 존재할 경우 기존 JobInstance를 리턴, 존재하지 않을 경우 새로운 JobInstance를 리턴
4. JobInstance가 생성되면 JobExecution도 새로 생성됨
5. 이미 존재하는 JobInstance에서 BatchStatus의 조건에 따라 Job의 재 실행 가능 여부 판단

 

예제 코드

@RequiredArgsConstructor
@Configuration
public class JobExecutionConfiguration {
 
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
 
    @Primary
    @Bean
    public Job executionJob() {
        return this.jobBuilderFactory.get("JobExecution")
                .start(executionStep1())
                .next(executionStep2())
                .build();
    }
 
    @Bean
    public Step executionStep1() {
        return stepBuilderFactory.get("JobExecutionStep1")
                .tasklet((contribution, chunkContext) -> {
                    JobExecution jobExecution = contribution.getStepExecution().getJobExecution();
                    System.out.println("jobExecution = " + jobExecution);
 
                    System.out.println("step1 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
 
    @Bean
    public Step executionStep2() {
        return stepBuilderFactory.get("JobExecutionStep2")
                .tasklet((contribution, chunkContext) -> {
//                    throw new RuntimeException("JobExecution has failed");
                    System.out.println("step2 has executed");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

위 코드를 작성하고 Batch Job을 실행해보면 정상적으로 실행되고 상태가 COMPLETED로 저장이 됩니다.

 

...

    @Bean
    public Step executionStep2() {
        return stepBuilderFactory.get("JobExecutionStep2")
                .tasklet((contribution, chunkContext) -> {
                    throw new RuntimeException("JobExecution has failed"); // 예외 발생
                })
                .build();
    }

위 코드처럼 step에서 강제로 예외를 발생시켜 보겠습니다.

 

이번에는 Batch Job에서 예외가 발생하여 상태가 FAILED로 기록이 됩니다.

 

만약 위 상태에서 동일한 JobParameters로 실행해보면 어떻게 될까요?

JobInstacne는 그대로지만 JobExecution는 다른 데이터가 저장되는 걸 확인할 수 있습니다.

따라서 JobInstance가 존재하더라도, Status가 FAILED일 경우 계속해서 Job을 실행할 수 있습니다. 또한 JobInstance와 JobExecution의 관계는 1 : N인 걸 확인할 수 있습니다.

 

위에서 강제로 예외 발생한 코드를 제거하고 Batch Job을 정상적으로 실행되도록 수정하여 테스트해보면 상태가 COMPLETED로 저장이 되고, 이후에는 동일한 JobInstance로 Job을 실행하는 건 불가능하게 됩니다.

 

 

 

참고 문서

• https://docs.spring.io/spring-batch/docs/current/reference/html/index-single.html
• https://www.inflearn.com/course/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B0%B0%EC%B9%98/dashboard