기억 파편

환경: spring batch 5.0.3, java 17

스프링 배치의 fault tolerant에는 크게 retry와 skip이 있다.

• skip(): ItemReader, ItemProcessor, ItemWriter 모두에서 예외 발생 시 적용할 수 있으며, 예외를 스킵하고 다음 아이템으로 넘어감
• retry(): ItemProcessor와 ItemWriter에만 적용되며, 예외 발생 시 설정된 횟수만큼 재시도. ItemReader에서는 retry()를 사용할 수 없음

이번 글에서는 writer의 retry에 대해 집중해 본다.

에러로 인한 재시도를 하고 싶을 경우. faultTolerant()와 함께. retry()와. retryLimit() 설정을 사용하면 Spring Batch에서 Step 또는 Chunk 단위로 처리 중 발생하는 예외에 대해 재시도 처리를 할 수 있다. 

private Step getCommonStep(
    JobRepository jobRepository,
    PlatformTransactionManager transactionManager,
    String stepName,
    MyBatisCursorItemReader<GmahjongRankingRat> itemReader,
    ItemWriter<GmahjongRankingRat> itemWriter) {
  return new StepBuilder(stepName, jobRepository)
      .listener(deleteAllBeforeStep)
      .<GmahjongRankingRat, GmahjongRankingRat>chunk(CHUNK_SIZE, transactionManager)
      .reader(itemReader)
      .writer(itemWriter) ///-> 에러 발생
      .faultTolerant()
      .retry(RuntimeException.class)
      .retryLimit(2)   // -> 재시도 두번
      .listener(retryListener())
      .build();
}

1. .faultTolerant()

• Fault Tolerant Step을 설정
• Step이나 Chunk 처리 중 예외가 발생했을 때, 해당 예외를 허용하거나 재시도할 수 있도록 구성
• retry, skip, noRollback, noRetry 등의 다양한 옵션을 적용할 수 있는 시작점

2. .retry(Class<? extends Throwable> exceptionClass)

• 재시도 대상 예외 타입을 지정
• 예외 타입은 특정 예외 클래스(예: RuntimeException.class) 또는 그 하위 클래스

예를 들어, .retry(RuntimeException.class)를 설정하면 RuntimeException과 그 하위 클래스에서 예외가 발생할 때마다 재시도가 이루어짐

3. .retryLimit(int retryLimit)

• 최대 재시도 횟수를 설정
• retryLimit 값은 예외가 발생했을 때 최대 재시도 가능 횟수를 의미(총 횟수가 아닌 재시도 횟수)
  • 예를 들어, retryLimit(2)로 설정하면 최대 2번 재시도함
• retryLimit에 설정된 횟수는 최초 시도에는 영향을 주지 않으며, 최초 시도 후 추가로 재시도할 수 있는 횟수를 의미

4. .faultTolerant().retry(RuntimeException.class).retryLimit(2)의 의미

• faultTolerant():
  • 예외가 발생했을 때, 예외를 처리하거나 재시도할 수 있는 내구성 있는 단계로 설정
• 재시도 대상 예외 설정 (retry(RuntimeException.class)):
  • RuntimeException과 그 하위 클래스에서 예외가 발생했을 때 재시도하도록 설정
• 최대 재시도 횟수 설정 (retryLimit(2)):
  • 최초 시도 외에 최대 2번의 재시도를 허용
  • 즉, 총 3번의 시도(초기 시도 1번 + 재시도 2번) <<<<<< 3일간 나를 고민에 빠트린 오늘의 주제...

재시도 테스트를 위해 3번 실행까지 에러를 발생시킨다.

  @Bean
  @StepScope
  public ItemWriter<GmahjongRankingRat> insertTotalRank(
      @Qualifier(DataSourceConfig.SESSION_FACTORY) SqlSessionFactory casualDb) {
    System.out.println(">>>>>>>>> insertTotalRank");
    return new ItemWriter<RankingRat>() {
      private final SqlSessionTemplate sqlSessionTemplate = new SqlSessionTemplate(casualDb);
      private int attempt = 0;

      @Override
      public void write(Chunk<? extends RankingRat> items) {
        attempt++;
        System.out.println("Attempt " + attempt + ": Writing items " + items);
        if (attempt < 3) { // 2번 이하로는 예외 발생
          throw new RuntimeException("Intentional error on attempt " + attempt);
        }
        for (RankingRat item : items) {
          sqlSessionTemplate.insert(
              Constant.GAME_MAPPER + "insertTotalRank", item);
        }
      }
    };
  }

retry 할 때 로그를 보기 위해 retryListener도 만들어서 달아준다.

  @Bean
  public RetryListener retryListener() {
    return new RetryListener() {
      @Override
      public <T, E extends Throwable> void onError(
          RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
        // 재시도 중 발생한 예외를 로깅
        System.err.println(
            "Retry attempt "
                + context.getRetryCount()
                + " failed with exception: "
                + throwable.getMessage());
      }
    };
  }

예상 시나리오

최초 시도 -> 에러1 -> 재시도1 -> 에러2 -> 재시도2 -> 에러3 -> 재시도 횟수 2번이 지나서 종료처리

그래서 에러가 3번까지 발생하고, 재시도 로그 2번 남을 것이라 기대

10:54:24.210 [main] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write:   ////최초 시도
10:54:24.221 [main] DEBUG o.s.batch.core.scope.StepScope - Creating object in scope=step, name=scopedTarget.insertTotalRank
>>>>>>>>> insertTotalRank
Attempt 1: Writing items... //최초 적재
...
10:54:24.247 [main] DEBUG o.s.b.core.step.tasklet.TaskletStep - Rollback for RuntimeException: java.lang.RuntimeException: Intentional error on attempt 1
10:54:24.247 [main] DEBUG o.s.t.support.TransactionTemplate - Initiating transaction rollback on application exception
java.lang.RuntimeException: Intentional error on attempt 1
...
10:54:24.250 [main] DEBUG c.a.i.imp.CompositeTransactionImp - rollback() done of transaction 10.78.130.172.tm172740206400000010
...
10:54:24.250 [main] DEBUG o.s.b.c.s.i.SimpleRetryExceptionHandler - Handled non-fatal exception
java.lang.RuntimeException: Intentional error on attempt 1
...
10:54:24.251 [main] DEBUG o.s.b.repeat.support.RepeatTemplate - Repeat operation about to start at count=2
...
Retry attempt 1 failed with exception: Intentional error on attempt 1


10:54:24.265 [main] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write: ////재시도 1
...
Attempt 2: Writing items... //재시도1 적재
10:54:24.282 [main] DEBUG o.s.b.core.step.tasklet.TaskletStep - Rollback for RuntimeException: java.lang.RuntimeException: Intentional error on attempt 2
10:54:24.282 [main] DEBUG o.s.t.support.TransactionTemplate - Initiating transaction rollback on application exception
java.lang.RuntimeException: Intentional error on attempt 2
...
10:54:24.282 [main] DEBUG c.a.i.imp.CompositeTransactionImp - rollback() done of transaction 10.78.130.172.tm172740206425100011
...
10:54:24.283 [main] DEBUG o.s.b.c.s.i.SimpleRetryExceptionHandler - Handled non-fatal exception
java.lang.RuntimeException: Intentional error on attempt 2
...
10:54:24.283 [main] DEBUG o.s.b.repeat.support.RepeatTemplate - Repeat operation about to start at count=3
10:54:24.288 [main] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write: ////재시도 2
...
Retry attempt 2 failed with exception: Intentional error on attempt 2


...
10:54:24.295 [main] DEBUG o.s.b.core.step.tasklet.TaskletStep - Rollback for RuntimeException: org.springframework.retry.ExhaustedRetryException: Retry exhausted after last attempt in recovery path, but exception is not skippable.
10:54:24.295 [main] DEBUG o.s.t.support.TransactionTemplate - Initiating transaction rollback on application exception
org.springframework.retry.ExhaustedRetryException: Retry exhausted after last attempt in recovery path, but exception is not skippable.
...
10:54:24.295 [main] DEBUG c.a.i.imp.CompositeTransactionImp - rollback() done of transaction 10.78.130.172.tm172740206428300012
...
10:54:24.298 [main] DEBUG o.s.b.repeat.support.RepeatTemplate - Handling fatal exception explicitly (rethrowing first of 1): org.springframework.retry.ExhaustedRetryException: Retry exhausted after last attempt in recovery path, but exception is not skippable.
10:54:24.299 [main] ERROR o.s.batch.core.step.AbstractStep - Encountered an error executing step gmahjongTotalRankingStep in job gmahjongDailyRankingJob

writer.write() 함수 두 번 호출됨. 3번째 시도를.. 하는 것 같긴 한데.. write 호출은 안 하고.. 횟수 초과로 전체 롤백을 하는 듯한 로그만 남음..

정확하게 세 번 시도를 하는 건지는 모르겠으나..(10:54:24.288과 10:54:24.295 사이에 실행 로그가 있어야 하지 않나)

"Rollback for RuntimeExceptioin", "rollback() done"의 로그가 세 번씩 남으므로 프로그램의 입장에선 에러를 세 번 감지한 것 같긴 하다..

참고로 rollback을 한다고 했지만 해당 chunk대한 부분만 rollback이라, beforeStep에서 한 deleteAll은 이미 적용되어 있고(테이블이 비어 있고) writer의 첫 chunk부터 에러 발생이라 결국 빈 테이블이 유지된다.

이 상태에서(retryLimit = 2; writer에서 첫 2번만 에러 발생)

retryLimit만 2 -> 3으로 올리면, 에러는 그대로 두 번, 총 시도는 최초 + 재시도 3번 = 4번이라 마지막에 성공해야 한다.

실행해보면, 위와 같은 에러가 두 번나고 마지막 쪽 로그가 아래처럼 바뀌면서 insert가 된다..

14:38:12 DEBUG [main] o.s.b.repeat.support.RepeatTemplate - Repeat operation about to start at count=3
14:38:12 DEBUG [main] o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write: ...
14:38:12 DEBUG [main] org.mybatis.spring.SqlSessionUtils - Creating a new SqlSession
...// insert logs

뜻대로 되긴 하는데.. 아직도 왜 마지막 시도에 대한 로그가 제대로 안 찍혔는지 모르겠다.. write을 세 번 부르는 게 아닌가?

테스트 코드로 확인

@Test
  void testRetryLimit() throws Exception {
    // 첫 번째, 두 번째, 세 번째 호출에서 예외 발생
    JobParameters jobParameters = getJobParameters();
    given(totalRankingReader.read()).willReturn(userRats().get(0), userRats().get(1), null);
    doNothing().doNothing().doNothing().when(deleteAllBeforeStep).beforeStep(any());
    doAnswer(
            invocation -> {
              System.out.println("First write attempt");
              throw new RuntimeException("error! 1");
            })
        .doAnswer(
            invocation -> {
              System.out.println("Second write attempt");
              throw new RuntimeException("error! 2");
            })
        .doAnswer(
            invocation -> {
              System.out.println("Third write attempt");
              throw new RuntimeException("error! 3");
            })
        .when(insertTotalRank)
        .write(any(Chunk.class));

    // Step 실행
    JobExecution jobExecution =
        jobTestUtils
            .getJobTester(GmahjongRankingJobConfig.JOB_NAME)
            .launchJob(jobTestUtils.makeJobParameters(jobParameters));

    verify(totalRankingReader, times(3)).read();
    // write 메서드가 총 3번 호출되었는지 확인
    verify(insertTotalRank, times(3)).write(any(Chunk.class));

    // Job이 실패했는지 확인 (최대 재시도 후 실패)
    assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.FAILED);
  }

retryLimit = 2 일 때 writer를 3번 부르는지 테스트 코드이다. 시도한 로그를 남기기 위해 doThrow가 아닌 doAnswer을 사용하였다.

위 테스트는 실패한다. 아래 부분에서 실제로 2번 호출했다고 검증된다.

verify(insertTotalRank, times(3)).write(any(Chunk.class));

관련 로그.. 본 로그와 크게 다르지 않다 세 번째 writer를 호출하는지 잘 모르겠는.. 로그다.

14:58:57.449 [Test worker] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write:
14:58:57.449 [Test worker] DEBUG o.s.retry.support.RetryTemplate - Retry: count=0
First write attempt
Retry attempt 1 failed with exception: error! 1
14:58:57.451 [Test worker] DEBUG o.s.retry.support.RetryTemplate - Checking for rethrow: count=1

14:58:57.459 [Test worker] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write:
14:58:57.460 [Test worker] DEBUG o.s.retry.support.RetryTemplate - Retry: count=1
Second write attempt
Retry attempt 2 failed with exception: error! 2
14:58:57.460 [Test worker] DEBUG o.s.retry.support.RetryTemplate - Checking for rethrow: count=2

14:58:57.466 [Test worker] DEBUG o.s.b.c.s.i.FaultTolerantChunkProcessor - Attempting to write:
14:58:57.466 [Test worker] DEBUG o.s.b.core.step.tasklet.TaskletStep - Rollback for RuntimeException: org.springframework.retry.ExhaustedRetryException: Retry exhausted after last attempt in recovery path, but exception is not skippable.
14:58:57.466 [Test worker] DEBUG o.s.t.support.TransactionTemplate - Initiating transaction rollback on application exception

파고들기..(w GPT)

1. retryLimit의 동작 원리

• retryLimit은 최대 재시도 가능 횟수를 의미합니다. 이는 "초기 시도 횟수를 제외한" 재시도 횟수입니다.
• 예를 들어, retryLimit = 2로 설정한 경우, 최초 시도 1번 + 재시도 2번 = 총 3번의 시도가 이루어집니다.
  • 시도 1: 최초 시도
  • 재시도 1: retryLimit에 의해 허용된 첫 번째 재시도
  • 재시도 2: retryLimit에 의해 허용된 마지막 재시도

2. 로그 분석

2.1. 첫 번째 시도 (count=0)

• First write attempt 로그가 나타난 후 RuntimeException이 발생하여 재시도가 필요합니다.
• retry()가 적용되어 재시도 1이 시작됩니다.

2.2. 첫 번째 재시도 (count=1)

• Second write attempt 로그가 나타난 후 두 번째 예외가 발생합니다.
• retry()에 의해 마지막 재시도가 필요합니다.

2.3. 두 번째 재시도 (count=2)

• 이 시도에서도 실패하여 RetryTemplate의 ExhaustedRetryException이 발생합니다.
• 세 번째 재시도에서 예외 발생 후, 더 이상 재시도가 이루어지지 않습니다.

3. 실제 호출과 예외 발생 횟수 차이

위 로그에서 write() 메서드가 총 3번 호출된 것을 확인할 수 있습니다. 하지만 예외가 명시적으로 두 번만 발생하는 이유는 다음과 같습니다:

1. 첫 번째 시도와 첫 번째 재시도: 첫 번째 시도와 첫 번째 재시도에서 예외가 발생하여 retry()가 작동했습니다. 이때 retryLimit에 따라 재시도가 시도됩니다.
2. 두 번째 재시도: 두 번째 재시도에서 예외가 발생하면 retryLimit을 모두 소진하게 됩니다. 이 시점에서 ExhaustedRetryException이 발생합니다.
3. 재시도 실패 후 처리: ExhaustedRetryException이 발생하면 더 이상 재시도가 이루어지지 않으며, 마지막 재시도 실패 후에는 예외가 발생한 상태로 처리됩니다.

4. ExhaustedRetryException의 의미

• ExhaustedRetryException은 설정된 재시도 한도(retryLimit)를 모두 소진한 후에도 예외가 해결되지 않았다는 의미입니다. 이 예외는 마지막 시도에서 예외가 발생했다는 것을 의미하며, 재시도가 더 이상 이루어지지 않음을 나타냅니다.

: 말이 애매하다. 총 3번의 시도를 하지만 writer를 부르는 시도는 아니고,, writer를 2번 부르고 재시도 횟수가 고갈되면 마지막 시도를 하기 전에 예외가 발생한다고 이해해야 할 것 같다.

참고로 첫 번째, 두 번째는 실패 세 번째에서 성공시키는 테스트 코드를 짜도 writer는 2번 불리고 전체 배치는 실패처리 된다.

최초 시도 -> 에러1 -> 재시도1 -> 에러2 -> 재시도2 -> 성공 -> ..전체 성공?이라고 생각하기 쉬운데..

이미 에러2에서 재시도 횟수(2)가 소비되어 에러가 발생하는 플로우다..

 @Test
  void testRetryLimit() throws Exception {
    JobParameters jobParameters = getJobParameters();
    given(totalRankingReader.read()).willReturn(userRats().get(0), userRats().get(1), null);
    doNothing().doNothing().doNothing().when(deleteAllBeforeStep).beforeStep(any());
    // 첫 번째, 두 번째 호출에서 예외 발생
	doAnswer(
            invocation -> {
              System.out.println("First write attempt");
              throw new RuntimeException("error! 1");
            })
        .doAnswer(
            invocation -> {
              System.out.println("Second write attempt");
              throw new RuntimeException("error! 2");
            })
        .doNothing()
        .when(insertTotalRank)
        .write(any(Chunk.class));

    // Step 실행
    JobExecution jobExecution =
        jobTestUtils
            .getJobTester(GmahjongRankingJobConfig.JOB_NAME)
            .launchJob(jobTestUtils.makeJobParameters(jobParameters));

    verify(totalRankingReader, times(3)).read();
    // write 메서드는 2번 호출
    verify(insertTotalRank, times(2)).write(any(Chunk.class));

    // Job이 실패했는지 확인 (최대 재시도 후 실패)
    assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.FAILED);

흐름 설명:

1. 첫 번째 시도 (Write attempt 1):
   • write() 호출 → RuntimeException 발생 → 재시도 1로 진입.
2. 첫 번째 재시도 (Write attempt 2):
   • write() 호출 → 다시 RuntimeException 발생 → 재시도 2로 진입.
3. 두 번째 재시도 (Write attempt 3):
   • write() 호출 → 성공적으로 처리됨.
4. 결과 처리:
   • 비록 세 번째 시도에서 성공했더라도, 재시도 한도인 retryLimit 2번을 모두 소진했으므로, 전체 Step이 실패로 처리됩니다.
   • ExhaustedRetryException이 발생하여, 더 이상의 재시도 없이 실패로 간주됩니다.

재시도 한도 소진과 처리 방식

1. 재시도 한도 소진:
   • retryLimit에 따라 재시도 횟수가 소진되면 RetryTemplate은 더 이상 ItemWriter를 호출하지 않습니다.
   • 재시도 중 실패한 예외를 처리할 수 없는 상태가 되면 ExhaustedRetryException을 발생시킵니다.
2. 최종 예외 처리:
   • 최종적으로 발생한 예외가 스킵 가능하지 않거나, 재시도 후에도 처리되지 않은 경우 전체 Step이 실패로 종료됩니다.
3. 마지막 시도가 성공하더라도:
   • 마지막 재시도에서 성공하더라도 재시도 한도가 모두 소진된 상태에서는 더 이상 재시도 없이 Step이 실패로 처리됩니다.

이.. 말싸움 때문에 3일 정도 매진한 것 같다..ㅠㅠ

결국, 재시도 횟수 차감되는 시점이 중요.....!

환경: springboot3.1.5, spring batch5, junit5

어찌어찌 배치 프로그램은 짰는데, 테스트코드는 어떻게 짜야할지 막막했다.

심지어 이 배치는 디비에서 오늘에 해당하는 데이터를 읽어 다른 디비에 적재하는 배치인데 

1. "오늘"이라는 날짜 디펜덴시가 있는 데이터가 필요하고
2. 이걸 타 디비에 실제로 넣어야 한다.

h2를 추가하여 로컬 배치로 돌리는 방법이 있겠지만 돌리는 날짜에 기반한 샘플 데이터를 만들어 넣는 게 좀 귀찮았고

디비 작업이야, 쿼리만 정확하면 보증되는 것이라(이미 다른 곳에서 돌고 있는 쿼리라서 실행이 보장되어 있음)

내가 검증하고 싶은 건 데이터를 정확히 꺼내오는 것이 아닌 job, step 등이 순차적으로 잘 도는지에 대해 작성하고 싶었다.

하여 db select, insert 부분을 mocking 할 수 있으면 좋겠다는 생각을 했다.

step1. get job launcher 

bean으로 등록하거나

(아래 코드 테스트 안 해봄)

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TestBatchConfig {

    @Bean
    public JobLauncherTestUtils jobLauncherTestUtils() {
        return new JobLauncherTestUtils();
    }
}

@SpringBootTest
@SpringBatchTest // mandatory?
@Import({TestBatchConfig.class, YourJobConfig.class})  // Replace YourJobConfig with your actual job configuration class
public class YourJobTest {

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

util로 만들어 빈으로 등록

public class JobTestUtils {

      @Autowired private ApplicationContext applicationContext;
      @Autowired private JobRepository jobRepository;
      @Autowired private JobExplorer jobExplorer;
      @Autowired private JobLauncher jobLauncher;

      public JobLauncherTestUtils getJobTester(String jobName) {
        Job bean = applicationContext.getBean(jobName, Job.class);
        JobLauncherTestUtils jobLauncherTestUtils = new JobLauncherTestUtils();
        jobLauncherTestUtils.setJobLauncher(jobLauncher);
        jobLauncherTestUtils.setJobRepository(jobRepository);
        jobLauncherTestUtils.setJob(bean);
        return jobLauncherTestUtils;
      }

      public JobParameters makeJobParameters(JobParameters parameters) {
        return new JobParametersBuilder(jobExplorer).addJobParameters(parameters).toJobParameters();
      }
      ...
  }

@TestConfiguration
public class TestBatchConfig {

  @Bean
  public JobTestUtils jobTestUtils() {
    return new JobTestUtils();
  }
}

@ActiveProfiles("test")
@Import({TestBatchConfig.class})
@SpringBootTest
class DailyRankingJobConfigTest {

  @Autowired private JobTestUtils jobTestUtils;

...
}

step2. mocking 하고자 하는 reader/writer가 빈으로 등록되어야 한다.

실제 job class에서 아래와 같이 item reader/writer가 주입되도록 하고..

@Configuration
@RequiredArgsConstructor
public class DailyRankingJobConfig {

  private final DailyRankingJobParameter jobParameter;

  @Qualifier("dailyRankingMatchCntReader")
  private final MyBatisCursorItemReader<Ranking> dailyRankingMatchCntReader;

  @Qualifier("dailyRankingGameMoneyReader")
  private final MyBatisCursorItemReader<Ranking> dailyRankingGameMoneyReader;

  @Qualifier("dailyRankingWriter")
  private final ItemWriter<Ranking> dailyRankingWriter;

테스트 코드에도 빈을 주입하는데.. @MockBean어노테이션을 이용한다. 여기서 주의할 건 name에 꼭 빈 이름을 넣어야 한다.. 안 그럼 못 찾는 듯.. 에러가 발생한다.

@ActiveProfiles("test")
@Import({TestBatchConfig.class})
@SpringBootTest
class DailyRankingJobConfigTest {

  @Autowired private JobTestUtils jobTestUtils;

  @MockBean(name = "dailyRankingMatchCntReader")
  private MyBatisCursorItemReader<Ranking> dailyRankingMatchCntReader;

  @MockBean(name = "dailyRankingGameMoneyReader")
  private MyBatisCursorItemReader<Ranking> dailyRankingGameMoneyReader;

  @MockBean(name = "dailyRankingWriter")
  private ItemWriter<Ranking> dailyRankingWriter;
  
  ...
  
   @Test
  @DisplayName("성공 케이스")
  void job__success() throws Exception {
    // given
    JobParameters parameters =
        new JobParametersBuilder()
            .addString(
                "date", LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd")), true)
            .addString("test version", UUID.randomUUID().toString(), true)
            .toJobParameters();

    given(dailyRankingMatchCntReader.read()).willReturn(getRanks().get(0), getRanks().get(1), null);
    given(dailyRankingGameMoneyReader.read()).willReturn(getRanks().get(1), null);
    doNothing().when(dailyRankingWriter).write(any());

    // when
    JobExecution jobExecution =
        jobTestUtils
            .getJobTester(DailyRankingJobConfig.JOB_NAME)
            .launchJob(jobTestUtils.makeJobParameters(parameters));

    // then
    assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
    // reader의 경우 chunk의 갯수만큼 호출
    verify(dailyRankingMatchCntReader, times(3)).read();
    verify(dailyRankingGameMoneyReader, times(2)).read();

    // writer의 경우 chunk 당 한번 호출(여기선 갯수가 적어 스텝 당 한 번임)
    final ArgumentCaptor<Chunk> captor = ArgumentCaptor.forClass(Chunk.class);
    verify(dailyRankingWriter, times(2)).write(captor.capture());
    List<Chunk> chunks = captor.getAllValues();
    assertThat(chunks.size()).isEqualTo(2);
    assertThat(chunks.get(0).size()).isEqualTo(2);
    assertThat(chunks.get(1).size()).isEqualTo(1);
  }

그러면 given.. willReturn/willThrow 등 기존에 사용하던 mocking 함수를 사용할 수 있게 된다!!

참고

https://jojoldu.tistory.com/236

환경: 자바 17, springboot3.1.5, springCloud 2022.0.4

목표: 배치가 하루에 한 번 돌아야 하고 (성공했어도) 종종 수동으로 한번 더 돌릴 수 있어야 함.

trial1: program argument로 date를 넘겨 중복 실행을 막아보자

step1. job parameter를 program argument로 넘겨야 한다.

시도한 방법

java -jar aaa.jar --job.name=sampleBatchJob dateParam=2022-09-09

에러 발생

Caused by: org.springframework.batch.core.converter.JobParametersConversionException: Unable to decode job parameter 2024-08-22
...
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ('-' (code 45)): Expected space separating root-level values
 at [Source: (String)"2024-08-22"; line: 1, column: 6]
...

그 어떤 글을 봐도 job param을 넘길 때 그냥 넘기길래.. 계속 저 방식으로 시도했지만.. 실패가 났다..

'-'가 문제 되나 싶어 지우고 해 봐도, 숫자가 아닌 임의의 문자열을 줘도 비슷한 에러가 나길래, 타입의 문제는 아닌 것 같았다.

step2. 혹시 springboot2 와 3의 차이로 인해 발생?

구글링 한 자료들이 outdated 된 것일 수 있다고 판단하였다.

그 이유는 springboot3 로 오면서 크게 변한 것 중 하나가 javax -> jakarta로 패키지명이 변한 것인데

사실 그 때는 jackson이라고 착각했다. 어쨌건 라이브러리 변화가 있어서 추가 설정이나 파라미터 넘기는 방식이 변했을지도 모르겠다고 생각했다.

그러다 구글링하다 파라미터에 type을 주는 예시를 봤는데, 아래와 같이 시도해 보았지만 역시나 같은 에러가 발생하였다.

dateParam(String)=2024-08-22

더 파고들어보니 위 방식은 fade out 되었고 boot3으로 버전이 오르면서 아래와 같이 바뀌었다는 글을 보게 된다.

parameter=value,type,identifying

그래서 아래와 같이 시도해 보았지만 여전히 실패하였다.

dateParam=2024-08-22,String,true

step3. 파고들기

관련 글을 좀 더 보다 보니 위와 같은 형태로 파라미터를 전달하려면 아래의 잡 파라미터 컨버터를 사용해야 한다고 한다. 이름 그대로 설정이 없을 경우 "기본적"으로 사용하는 컨버터이다.

DefaultJobParametersConverter

해당 프로젝트의 컨버터 설정이 뭔지 찾아보니 맙소사.. 다른 것이었다.

 @Bean
  public JobParametersConverter jobParametersConverter() {
    return new JsonJobParametersConverter();
  }

해당 컨버터를 사용할 경우 잡 파라미터를 아래와 같은 형태로 넘겨야 한다고 한다.

parameterName='{"value": "parameterValue", "type":"parameterType", "identifying": "booleanValue"}'

그래서 비슷하게 만들고 실행해 본다.

--job.name=sampleBatchJob
dateParam='{"value":"2024-08-22","type":"java.lang.String","identifying":"true"}'

아래의 에러가 발생한다.

Caused by: org.springframework.batch.core.converter.JobParametersConversionException: Unable to decode job parameter '{value:2024-08-22,type:java.lang.String,identifying:true}'
...
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character (''' (code 39)): expected a valid value (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
...

json parsing 에러다. single quotaion, double quotation 문제인가 싶어서 여러 조합으로 수정해 봤는데도 비슷한 에러만 발생한다.

그러던 중 한 글을 보게 되는데, json 안의 quote에는 escape 처리를 해주어야 한다는 것! (https://github.com/spring-projects/spring-batch/issues/4299)

그래서 아래처럼 수정했더니 드디어 돌아간다!

--spring.profiles.active=local
--job.name=sampleBatchJob
dateParam="{\"value\":\"2024-08-22\",\"type\":\"java.lang.String\",\"identifying\":\"true\"}

복수개의 파라미터를 넘기게 된다면 아래와 같다. 필요없는 콤마, 따옴표.. 등등이 들어가면 뜬금없는 에러가 나며 인식이 되지 않는다(에러 상황을 알기 어려움).

--spring.profiles.active=local
--job.name=DailyRankingJob
date="{\"value\":\"2024-08-21\",\"type\":\"java.lang.String\",\"identifying\":\"true\"}"
version="{\"value\":\"1\",\"type\":\"java.lang.Integer\",\"identifying\":\"true\"}"

json job converter 예시: https://spring.io/blog/2022/11/24/spring-batch-5-0-goes-ga

배치 최신 문서: https://docs.spring.io/spring-batch/reference/job/running.html

trial2. 날짜는 넘겼는데, 잡에서 사용하게 해야 하네!

job parameter를 프로젝트에서 보이게 하려면 우선 빈으로 등록되어 있어야 한다.

아래와 같이 일반적인 string으로 받을 경우 아래의 에러를 만난다.

  @Value("#{jobParameters[dateParam]}")
  public String dateParam;

Caused by: org.springframework.expression.spel.SpelEvaluationException: 
	EL1008E: Property or field 'jobParameters' cannot be found on object of type 'org.springframework.beans.factory.config.BeanExpressionContext' - maybe not public or not valid?

빈만 등록되면 안 되고 scope에서 보이게끔 선언해야 한다.

Job 안에서 보이게 하려면 JobScope, Step 안에서 보이려면 StepScope 안에서 사용하게 끔 아래와 같이 빈 선언부에 등록한다.

@Bean(STEP_NAME)
@JobScope
public Step rankingStep(
    JobRepository jobRepository,
    PlatformTransactionManager transactionManager,
    MyBatisCursorItemReader<Ranking> dailyRankingReader,
    ItemWriter<Ranking> dailyRankingWriter,
    @Value("#{jobParameters[dateParam]}") String dataParam) {
    	...
    }

이때 Job Parameter의 타입으로 사용할 수 있는 것으로는 Double, Long, Date, String이 있다.(배치4 기준)

LocalDate나 LocalDateTime같은 타입은 String으로 받아서 타입 변환을 해야 한다. 반환하는 방법은 크게 세 가지가 있다(https://jojoldu.tistory.com/490). 여기서는 setter주입 방식으로 해본다.

@Getter
@NoArgsConstructor
@Component
@JobScope
public class DailyRankingJobParameter {
  private LocalDate date;

  @Value("#{jobParameters[dateParam]}")
  public void setDate(String date) {
    this.date = LocalDate.parse(date, DateTimeFormatter.ofPattern("yyyy-MM-dd"));
  }
}

빈으로 등록되어야 JobScope이 먹기 때문에 굳이? 싶어도 Component 등록을 해줘야 한다.

사용하고자 하는 Job이나 Step에서는 argument로 전달할 필요 없이 클래스에서 생성자로 받아서 바로 사용하면 된다.

@Configuration
@RequiredArgsConstructor
public class DailyRankingJobConfig {

	private final DailyRankingJobParameter jobParameter;
    
    ...
    
  @Bean(STEP_NAME)
  @JobScope
  public Step sinyutnoriDailyRankingStep(
      JobRepository jobRepository,
      PlatformTransactionManager transactionManager,
      MyBatisCursorItemReader<SinyutnoriRanking> dailyRankingReader,
      ItemWriter<SinyutnoriRanking> dailyRankingWriter
      //      @Value("#{jobParameters[dateParam]}") String dataParam
      ) {
    System.out.println(jobParameter);
    return new StepBuilder(STEP_NAME, jobRepository)
        .<SinyutnoriRanking, SinyutnoriRanking>chunk(CHUNK_SIZE, transactionManager)
        .reader(dailyRankingReader)
        .writer(dailyRankingWriter)
        .build();
  }

위 내용은 fade out된 내용이고(물론 위처럼 해도 작동은 됨) 실제로는 job parameter class를 만들 필요도 없이! argument에 아래와 같이 전달하면 된다.

--job.name=DailyRankingJob
date="{\"value\":\"2024-08-21\",\"type\":\"java.time.LocalDate\",\"identifying\":\"true\"}"
version="{\"value\":\"2\",\"type\":\"java.lang.Integer\",\"identifying\":\"true\"}"

사용하려는 job/step에서 바로 땡겨다 사용 가능. 클래스에 선언하면 scope이 정의되지 않아 에러가 난다.

@Bean(STEP1_NAME)
@JobScope
public Step DailyRankingMatchCntStep(
    JobRepository jobRepository,
    PlatformTransactionManager transactionManager,
    @Value("#{jobParameters[date]}") LocalDate date) {

In Spring Batch 5, when job parameters are passed as strings, Spring Batch will automatically infer the correct type (
String, Long, Double, or Date) based on the format of the input.

trial3. 날짜는 같은데도 반복 실행이 된다?

위에서 날짜를 받아서 job parameter로 넘기는 것을 해봤다. 근데도 여전히 반복 실행이 된다. 왜 그런가 싶어 job execution에 사용된 파라미터를 확인해 보니 아래와 같이 run.id와 복합 키로 잡고 있어서 매번 다르게 인식하고 있었다.

해당 부분은 소스로 보면 아래와 같은데, RunIdIncrementer가 run.id의 키로 하나씩 키를 증가시키면서 실행하기 때문이다.

  @Bean(JOB_NAME)
  public Job rankingJob(JobRepository jobRepository, Step rankingStep) {
    return new JobBuilder(JOB_NAME, jobRepository)
        .incrementer(new RunIdIncrementer())  // <----
        .start(rankingStep)
        .build();
  }

따라서 해당 부분을 주석하면 여러 번 실행되지 않는 것을 확인할 수 있다.

16:32:21.071 [main] ERROR o.s.boot.SpringApplication - Application run failed
java.lang.IllegalStateException: Failed to execute ApplicationRunner
...
Caused by: org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException: A job instance already exists and is complete for parameters={'dateParam':'{value=2024-08-21, type=class java.lang.String, identifying=true}'}.  If you want to run this job again, change the parameters.

체크포인트

• RunIdIncrementer와 같이 매번 다른 키를 생성하는 job incrementer를 사용하지 않았는지 확인
• argument로 파라미터를 넘길 때 identifying 값을 true로 넘겼는지 확인
  • 해당 의미는 고유 키값인지 의미로 true면 execution key로 인식한다.

dateParam="{\"value\":\"2024-08-21\",\"type\":\"java.lang.String\",\"identifying\":\"true\"}

• 이전 execution 결과, params 확인
  • 이전 execution이 fail이면 execution id는 달라도 같은 job instance의 execution으로 묶여 재실행이 가능하다.
  • 해당 내용을 확인할 수 있는 쿼리 참고..

select execution.JOB_EXECUTION_ID, execution.JOB_INSTANCE_ID, execution.CREATE_TIME, execution.STATUS, execution.EXIT_CODE, execution.EXIT_MESSAGE,
params.PARAMETER_NAME, params.PARAMETER_TYPE, params.PARAMETER_VALUE, params.IDENTIFYING
FROM BATCH_CASUAL_JOB_EXECUTION execution inner join BATCH_CASUAL_JOB_EXECUTION_PARAMS params on execution.JOB_EXECUTION_ID = params.JOB_EXECUTION_ID 
order by JOB_EXECUTION_ID  DESC
;

job instance 45의 경우 dateParam, identifying: true로 실패 -> 성공을 했고(execution 47, 48)

job instance 46번의 경우, 같은 dateParam이지만 identifying:false로 실행을 하니 다른 job parameter로 인식을 해서 실행을 되었고 실패 난 것을 알 수 있다.

결론

배치가 하루에 한 번 돌아야 하고 종종 수동으로 한번 더 돌릴 수 있으려면..

배치 당 unique 한 값(날짜 등)을 argument로 넘기고 job parameter로 받아서 적당한 scope의 빈에 등록해야 한다.

나의 경우, json의 형식으로 parameter를 넘기고 혹시 재실행이 필요할 경우 identifying: false로 재실행하려고 한다.

참고

argument로 파라미터 보내는 방법

https://velog.io/@guswns3371/Spring-Boot-Framework-%EB%B2%84%EC%A0%84-%EC%97%85%EA%B7%B8%EB%A0%88%EC%9D%B4%EB%93%9C-%EA%B3%BC%EC%A0%95

파라미터를 bean으로 등록한다는 것의 의미

https://velog.io/@lxxjn0/Spring-Batch-Guide-05.-Spring-Batch-Scope-Job-Parameter

batch5에 추가된 내용

https://devfunny.tistory.com/931

환경: springboot3, spring batch5, mybatis

그동안 jpa만 주구장창 사용했어서 올만에 Mybatis 설정이다!

1. 디비 정보 등록(application.yml)

2. 빈 등록

@Configuration
@RequiredArgsConstructor
@MapperScan(
    value = {"com.batch.ranking.mapper"},
    annotationClass = LogDataSource.class,
    sqlSessionFactoryRef = "LogDbSqlSessionFactory",
    sqlSessionTemplateRef = "LogDbSqlSessionTemplate")
public class LogDataSourceConfig {

  public static final String SOURCE_DATASOURCE_NAME = "LogDbDataSource";

  @Value("classpath:mybatisConfig.xml")
  private Resource configLocation;

  @Bean(SOURCE_DATASOURCE_NAME)
  public DataSource LogDbDataSource() {
    DataSourceProperty dataSourceProperty = //get them from property

    Properties properties = new Properties();
    properties.setProperty("url", dataSourceProperty.getJdbcUrl());
    properties.setProperty("user", dataSourceProperty.getUsername());
    properties.setProperty("password", dataSourceProperty.getPassword());

    AtomikosDataSourceBean dataSource = new AtomikosDataSourceBean();
    dataSource.setUniqueResourceName(SOURCE_DATASOURCE_NAME);
    dataSource.setXaDataSourceClassName("com.mysql.cj.jdbc.MysqlXADataSource");
    dataSource.setXaProperties(properties);
    dataSource.setMaxPoolSize(connectionPoolProperty.getMaximumPoolSize()); //from property
    dataSource.setMinPoolSize(connectionPoolProperty.getMinimumIdle());

    return dataSource;
  }

//Qualifier is mandatory otherwise it will connect to Primary bean
  @Bean
  public SqlSessionFactory LogDbSqlSessionFactory(
      @Qualifier("LogDbDataSource") DataSource LogDbDataSource) throws Exception {
    SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
    bean.setConfigLocation(configLocation);
    bean.setDataSource(LogDbDataSource);
    return bean.getObject();
  }

  @Bean
  public SqlSessionTemplate LogDbSqlSessionTemplate(
      @Qualifier("LogDbSqlSessionFactory") SqlSessionFactory LogDbSqlSessionFactory) {
    return new SqlSessionTemplate(LogDbSqlSessionFactory);
  }
}

2-1. mybatis 설정은 자바로 해도 되지만 분리하는 게 가독성이 좋아서 분리하였다.

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0/EN"
        "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
    <settings>
        <setting name="mapUnderscoreToCamelCase" value="true"/>
    </settings>

    <typeAliases>
        <typeAlias type="com.batch.adapter.mybatis.handlers.RankingTypeHandler" alias="RankingTypeHandler" />
    </typeAliases>
</configuration>

2-2. 매퍼 마킹하는 어노테이션

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
public @interface LogDataSource {}

3. 사용하는 job에서 부를 경우

  private final SqlSessionFactory LogDbSqlSessionFactory;

  @Bean
  @StepScope
  public MyBatisCursorItemReader<Ranking> DailyRankingReader() {
    return new MyBatisCursorItemReaderBuilder<Ranking>()
        .sqlSessionFactory(LogDbSqlSessionFactory)
        .queryId(
            "com.batch.domain.ranking.mapper.DailyRankMapper.selectDailyTop100")
        .build();
  }

4. 매퍼에 쿼리 작성

@LogDataSource
public interface DailyRankMapper {

  List<Ranking> selectDailyTop100();
}

<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.batch.domain.ranking.mapper.DailyRankMapper">
  <resultMap id="sinyutnoriRanking"
    type="com.batch.domain.ranking.model.Ranking">
    <constructor>
      <idArg column="memberid" javaType="java.lang.String" name="memberId"/>
      <arg column="regdate" javaType="java.time.LocalDate" name="registerDate"/>
      <arg column="kind" javaType="com.batch.domain.ranking.type.RankingType" name="RankingType" typeHandler="RankingTypeHandler"/>
      <arg column="gamemoney" javaType="java.lang.Long" name="gameMoney"/>
      <arg column="winrate" javaType="java.lang.Long" name="winRate"/>
      <arg column="matchcnt" javaType="java.lang.Long" name="matchCount"/>
      <arg column="wincnt" javaType="java.lang.Long" name="winCount"/>
      <arg column="defeatcnt" javaType="java.lang.Long" name="defeatCount"/>
      <arg column="ranking" javaType="java.lang.Integer" name="ranking"/>
    </constructor>
  </resultMap>

  <select id="selectDailyTop100" resultMap="Ranking">
    <![CDATA[
    SELECT  memberid
         , regdate
         , kind
         , gamemoney
         , winrate
         , matchcnt
         , wincnt
         , defeatcnt
         , @RNUM := @RNUM + 1 AS ranking
    FROM Table
      ORDER BY (wincnt + defeatcnt + drawcnt) DESC
      , (wincnt / (wincnt + defeatcnt + drawcnt)) * 100 DESC
      , gamemoney
      ) B, (SELECT @RNUM := 0) r
    WHERE @RNUM < 100
    ]]>
  </select>
</mapper>

5. enum으로 바로 꺼내고 싶다면 type handler 작성

public class RankingTypeHandler extends BaseTypeHandler<RankingType> {

  @Override
  public void setNonNullParameter(
      PreparedStatement ps, int i, RankingType parameter, JdbcType jdbcType)
      throws SQLException {
    ps.setInt(i, parameter.getCode());
  }

  @Override
  public RankingType getNullableResult(ResultSet rs, String columnName)
      throws SQLException {
    return RankingType.findByCode(rs.getInt(columnName)).orElse(null);
  }

  @Override
  public RankingType getNullableResult(ResultSet rs, int columnIndex)
      throws SQLException {
    return RankingType.findByCode(rs.getInt(columnIndex)).orElse(null);
  }

  @Override
  public RankingType getNullableResult(CallableStatement cs, int columnIndex)
      throws SQLException {
    return RankingType.findByCode(cs.getInt(columnIndex)).orElse(null);
  }
}

여기서 궁금한 사항

Mapper interface를 직접 사용하지 않은데도(자바 로직에서) 클래스가 필요한 것인가?

reader/writer를 보면 아래와 같이 직접 db factory를 연결했으니 interface는 필요 없는 게 아니냐는 문의를 주셔서 좀 더 알아본다.

  @Bean
  @StepScope
  public MyBatisCursorItemReader<SinyutnoriRanking> dailyRankingMatchCntReader(
      @Qualifier(HangameLogDataSourceConfig.SESSION_FACTORY) SqlSessionFactory logDb) {
    return new MyBatisCursorItemReaderBuilder<SinyutnoriRanking>()
        .sqlSessionFactory(logDb)
        .queryId(LOG_MAPPER + "selectDailyTop1000UsersByMatchCnt")
        .build();
  }

설정이 대략 이런식으로 연결되어 있다고 할 때...

@Configuration
@RequiredArgsConstructor
@MapperScan(
    value = {"com.batch.domain.mapper.gamemapper.*"},
    annotationClass = DataSource.class,         //mapper interface에 해당 어노테이션을 달아야 
    sqlSessionFactoryRef = DataSourceConfig.SESSION_FACTORY,
    sqlSessionTemplateRef = "DbSqlSessionTemplate")
public class DataSourceConfig {

@DataSource
public interface GameMapper {

<mapper namespace="com.batch.domain.mapper.gamemapper.GameMapper">

1. interface 삭제 가능? No

:  GameMapper삭제하고 (Config에 annotationClass 주석하니; 하건 안하건 둘 다) 에러 발생

Caused by: java.lang.IllegalArgumentException: Mapped Statements collection does not contain value for com.batch.domain.mapper.gamemapper.GameMapper.insertRank

xml의 namespace가 인터페이스와 연결되어 있어야 쿼리 주입이 가능

2. 직접적인 함수의 호출이 없으므로 함수는 삭제 가능? Yes

xml안에는 <select>, <insert> 등 여러 쿼리가 있지만 직접 호출하지 않으므로 interface에 연결하는 함수는 없어도 된다.

@DataSource
public interface GameMapper {

//  List<Ranking> selectDailyTop1000UsersByMatchCnt();
}

@Bean
@StepScope
public MyBatisCursorItemReader<Ranking> dailyRankingMatchCntReader(
    @Qualifier(DataSourceConfig.SESSION_FACTORY) SqlSessionFactory logDb) {
  return new MyBatisCursorItemReaderBuilder<Ranking>()
      .sqlSessionFactory(logDb)
      .queryId(LOG_MAPPER + "selectDailyTop1000UsersByMatchCnt")
      .build();
}

select 함수가 직접적으로 선언되지 않아도 작동한다.

작동은 되지만 나중에 관리차원에서 헷갈릴까 봐 지울지 말지 약간 걱정은 된다..

Spring Batch 소개

• Accenture 와 Pivotal이 협업하여 개발한 배치 프레임워크.
• 정보를 가공하여 통계를 생성 시
• 매우 큰 데이터를 주기적으로 처리해야 할 때
  • 일반적으로 전체 로그 데이터를 수집하고 통합하는 과정이 필요함.
  • e.g : DAU, WAU, MAU, WoW, QoQ, YoY
    • DAU, WAU, MAU : access log 데이터 추출 + unique 유저 계산
• 내부/외부 시스템으로부터 생산된 다양한 형태의 데이터를 통합해야 할 때
  • 두개 이상의 데이터를 원하는 형태로 가공한 뒤, 조합하여 원하는 데이터를 생산.
  • e.g. CTR(클릭률 : Click Through Rate), CPC (클릭당 비용 : Cost Per Click), CVR(전환율 : Conversion Rate)
    • CTR == 클릭수 / 노출수 == access log 데이터 추출
    • CPC == 광고 집행비용 / 클릭수 == 비용 테이블과 access log 데이터 추출
    • CVR == 전환 수 / 클릭수 == 전환 계산에 필요한 로그 또는 테이블 + access log 데이터.
• 배치 프로그래밍?
  • 배치 프로그램은 여러개의 작업으로 구성
  • 작업은 어러개의 단계로 구성
  • c.f. 젠킨스 파이프 라이닝, 데이터 파이프 라인, Airflow DAG Pipeline
    
    • A -> B -> C
    • 추상화; 플로우 강제; 등의 framework
  • 같은 단계들의 반복.

Why Spring Batch? - 기술적 목표

• 스프링 배치 프레임워크의 기능을 사용하여 비지니스 로직에 집중한다.
• 즉시 사용할 수 있는 (Out-of-box) 실행 인터페이스를 제공
• 인프라 계층(DB Reader, Queue reader ..)과 구분되어 있음.

Why Spring Batch?

• 스프링 배치(Spring Batch)는 스케줄링 프레임워크가 아님.
• 스케줄링은 다른 도움을 받아서 해야한다..
  
  • 스케줄링 프레임워크 : Quartz, Tivoli, Control-M, and others
  • 스프링 프레임워크 : @Scheduled
  • 리눅스의 Crontab
  • Jenkins, Rundeck etc.
  • 스프링 배치 프레임워크 애플리케이션은 스케줄러와 함께 작동하도록 설계됨.
• 스프링 프레임워크가 제공하는 기능들
  • Transaction management
  • Chunk based processing
  • Start/Stop/Restart
  • Retry/Skip
  • Job 처리 통계
  • Web based administration interface (Spring Cloud Data Flow)
    • 따로 모듈이 있다.

JobBuilderFactory > JobBuilder > SimpleJobBuilder > SimpleJob

• validate: 로직 전 job parameter 검증 가능
  
  • JobParametersValidator implement 해서 커스텀하게 만들 수 있음

DefaultJobParametersValidator(String[] requiredKeys, String[] optionalKeys)

• prevent: job의 재시작 여부 설정
  • 기본값은 true이며 false일 경우 이 job은 재시작을 지원하지 않는다는 의미 -> 재시작하려고 하면 exception발생
  • 첫 시작과는 무관
  • job의 성공/실패와 상관없이 오직 preventRestart 설정 값에 따라 실행 여부를 판단

• incrementer: jobParameters에 필요한 값을 증가시켜 다음에 사용될 jobParameters 리턴
  • 기존의 jobParameter 변경 없이, 이전에 실패하지 않았더라도, job을 여러번 시작하고자 할 때(ex. 검사하는 로직 등)
  • 사용하지 않는 파라미터를 추가, 그 값을 변경시켜 마치 다른 파라미터처럼 보이게 함(인덱스를 추가해서 ++시킨다던가, 현재 날짜를 추가한다거나)
  • RunIdIncrementer implement해서 커스텀하게 만들 수 있음

<SimpleJob>

spring-boot 2.7.0 기준 작성

• 특정 job만 실행할 수 있는 옵션

application.yml에 spring.batch라는 prefix로 설정해 두면 BatchProperties 라는 파일에서 읽어감(,로 복수개 구분)

spring:  
  batch:
    job:
      names: ${job.name:NONE} 

# program argument에 --job.name의 옵션으로 준 이름을 받아와서 실행, 
# 해당 이름의 job 없으면 NONE이라는 이름의 배치 실행; 없으면 실행안함

• 부트 실행 시 자동 실행 막는 옵션

spring:
  batch:	
    job:
      enabled: false
      
# 기본 값 true

• 디비 스키마 관련 옵션

spring:
  batch:
    jdbc:
      initialize-schema: always
      table-prefix: ST_
      
# initialize-schema
#   ALWAYS : 항상 실행(없으면 생성)
#   EMBEDDED : 내장 DB일 때만 실행
#   NEVER : 항상 실행 안함

# table-prefix 테이블 프리픽스 변경(기본: BATCH_)
# 이 때 테이블을 미리 생성해주어야 한다. 그 이름 테이블이 없다고 다시 만들지는 않음.

1. h2 설치

https://inma.tistory.com/149

https://so-easy-coding.tistory.com/5?category=968591 

2. spring-batch 에 필요한 테이블들

•  spring-batch-core/org.springframework/batch/core/* 에 위치

• 스크립트 실행 설정은 application.properties의 spring.batch.initialize-schema 로 구분
  • ALWAYS : 항상 실행
  • EMBEDDED : 내장 DB일 때만 실행 (기본 값)
  • NEVER : 항상 실행 안 함

3. springboot 와 h2 연결

spring:
  datasource:
    hikari:
      jdbc-url: jdbc:h2:tcp://localhost/~/spring-batch-test
      username: sa
      password:
      driver-class-name: org.h2.Driver
  batch:
    job:
      enabled: false #구동 시 자동실행 안함
    jdbc:
      #ALWAYS : 항상 실행
      #EMBEDDED : 내장 DB일 때만 실행
      #NEVER : 항상 실행 안함
      initialize-schema: embedded

참고) h2는 다양한 모드를 제공

https://www.h2database.com/html/features.html

• 엠비디드 연결 : jdbc:h2:[file:][<path>][databaseName]
• 인 메모리 : jdbc:h2:mem:<databaseName>
• 서버 모드 : jdbc:h2:tcp://<server>[:<port>]/[<path>]<databaseName>

intellij 안의 database로 테이블을 보는게 더 편할 것 같아서 설정했는데 connection이 성공하는 듯 하다가 아래와 같이 실패하는 현상이 있다.

https://youtrack.jetbrains.com/issue/DBE-15020

뭔가 intelij 버그 같아서 기다려야 할 것 같다.

Job

• 가장 상위 개념, 하나의 배치작업 자체
• 전체적으로 설정하고 명세한 객체
• 여러 step을 포함하고 있는 컨테이너, 반드시 한 개 이상의 step으로 구성해야 함
• 최상위 인터페이스
• 기본 구현체:
  • SimpleJob: 순차적으로 step을 실행시키는 job. 모든 job에서 유용하게 사용할 수 있는 표준 기능을 가지고 있음. steps 리스트 안에 step 객체를 가지고 있음
  • FlowJob: 특정한 조건과 흐름에 따라 step을 구성하여 실행시킴, 더 유연, 조건과 흐름에 따라 구성을 다르게 할 수 있음. flow 객체를 실행히켜서 작업을 진행

JobInstance

• job이 실행될 때 만들어지는 논리적 실행 단위
• job의 설정과 구성은 동일하지만 job이 실행되는 시점에 처리하는 내용(jobParameter)은 다르기 때문에 job의 실행을 구분해야 함
• job : jobInstance = 1 : M

JobParameter

• job 실행 시 함께 포함하여 사용하는 파라미터 객체
• 하나의 job에 존재할 수 있는 여러 jobInstance를 구분하기 위한 용도
• jobParameter : jobInstance = 1 : 1
• STRING, DATE, LONG, DOUBLE 타입 제공

바인딩 방법

• jar 실행 시 주입
  • java -jar dd.jar requestDate=111
• 코드로 생성
  • JobParameterBuilder, DefaultJobParametersConverter
• SpEL 이용
  • @Value("#{jobParameter[requestDate]}")

참고) jar 실행 시 주입 하는 방법으로 테스트 해봤더니 아래와 같은 오류가 난다. 

libs % java -jar springbatch-test-0.0.1-SNAPSHOT.jar name=user3 seq(long)=1L date(date)=2021/02/03 age(double)=12.34
zsh: no matches found: seq(long)=1L

java -jar batch.jar executionDate\(date\)=2021/02/21
or
java -jar batch.jar 'executionDate(date)=2021/02/21'

JobExecution

• jobInstance에 대한 한 번의 시도를 의미하는 객체. job이 실행될 때 마다 생성되며 job 실행 중에 발생한 정보를 저장하는 객체
• jobExecution은 FAILED or COMPLETED 등의 job 실행 결과를 가지고 있음
• COMPLETED가 될 때 까지 하나의 jobInstance 내에서 여러번 시도가 생길 수 있음
• JobInstance : JobExecution = 1 : M

Step

• job을 구성하는 독립적인 하나의 단계로 실제 배치 처리를 정의하고 컨트롤하는데 필요한 모든 정보를 가진 객체
• 배치 작업을 어떻게 구성하고 실행할지 세부작업을 task기반으로 설정하고 명세해 놓은 객체
• 비지니스 로직을 담음
• 기본 구현체 
  • taskletStep : 기본이 되는 클래스, 구현체 제어
  • partitionStep : 멀티 스레드 방식, step을 여러 개로 분리해서 실행
  • jobStep : step 내에서 job을 실행하도록 함, chaining
  • flowStep : step 내에서 flow를 실행하게 함

StepExecution

• step에 대한 한 번의 시도를 의미하는 객체, step 실행 중에 발생한 정보를 저장하는 객체
• 각 step별로 생성, step이 실제로 시작되었을 때만 생성
• job이 실패해 재시작하더라도 이미 성공한 step은 재실행되지 않고 실패한 step만 실행됨(실패하면 그 이후 step은 실행되지 않고 종료)
• 모든 stepExecution이 성공해야 jobExecution도 정상적으로 완료
• JobExecution : StepExecution =  1 : M

StepContribution

• 청크 프로세스의 변경사항을 버퍼링한 수 stepExecution상태를 업데이트하는 도메인 객체
• 청크 커밋 직전에 stepExecution의 apply 메서드를 호출하여 상태를 업데이트
• exitStatus의 기본 종료토드 외 사용자 정릐 종료코드를 생성해서 적용할 수 있음
• stepExecution 객체 안에 있음

ExecutionContext

• map의 형태로 stepExecution 이나 jobExecution 객체의 상태를 저장하는 공유 객체; 유지 관리에 필요한 키-값 저장
• DB에는 직렬화 한 값으로 저장됨({"키":"값"})
• 공유 범위
  • step: 각 step의 stepExecution에 저장되며 step간 공유 안 됨
  • job: 각 job의 jobExecution에 저장되며 job간 공유 안되고 job의 step간에는 서로 공유 됨
    • job 재시작 시 이미 처리한 데이터는 건너뛰고 이후를 수행하도록할 때 상태정 보를 활용 

JobRepository

• 배치 작업 중 정보를 저장하는 저장소, 인터페이스; simpleJobRepository 구현체 있고, 커스텀 가능
• job이 언제 수행되었고 언제 끝났고 몇 번 실행되었고 등 메타데이터를 저장함
• @EnableBatchProcessing 어노테이션만 선언하면 job repository가 자동으로 빈으로 생성됨
• jdbc 방식
  • JobRepostioryFactoryBean
  • 내부적으로 aop를 통해 트랜잭션 처리 해주고 있음
  • 트랜잭션 isolation 기본 값은 serializable로 최고 수준이지만 다른 레벨로도 지정 가능
  • 메타테이블 테이블 prefix가 기본으로는 BATCH_ 이지만 수정 가능
• in memory 방식
  • MapJobRepositoryFactoryBean
  • 성능 등의 이유로 굳이 DB에 넣고 싶지 않은 경우
  • test나 프로토타입의 빠른 개발이 필요할 때 사용

JobLauncher

• job을 실행하는 역할
• job과 JobParameter를 인자로 받아 요청된 배치 작업을 수행한 후 jobExecution을 반환
• 부트가 구동되면 jobLauncher 빈이 자동 생성됨
• job 실행
  • 동기적 실행: SyncTaskExecutor
    • 모든 단계를 다 마친 후에 jobExecution 반환
    • 스케줄러에 의한 배치 처리에 적합
    • 배치 처리 시간이 길어도 상관 없을 경우
  • 비동기적 실행: SimpleAsyncTaskExecutor
    • jobExecution을 생성하고 획득, 획득하자마자 반환(exitStauts.UNKNOWN) , 배치 완료
    • http 요청에 의한 배치처리에 적합; 배치처리 시간이 길 경우 응답이 늦어지지 않도록 함
     

<배치 패턴>

• read: 데이터 조회
• process: 데이터 가공
• write: 수정된 양식으로 다시 저장

<아키텍쳐>

• 개발자가 만든 모든 배치 job과 커스텀 코드 포함
• 개발자는 로직의 구현에만 집충, 공통 기반기술은 프레임워크가 담당

• job을 실행, 모니터링, 관리하는 api로 구성

• application, core 모두 공통 Infrastructure 위에서 빌드
• job실행의 흐름과 처리를 위한 틀 제공
• reader, processor, writer, skip, retry 등

<배치 활성화>

@EnableBatchProcessing  어노테이션을 달아야 실행되며, 아래 클래스들이 활성화 됨

• BatchAutoConfiguration -> JobLauncherApplicationRunner 빈 생성
• SimpleBatchConfiguration -> 스프링 배치 주요 구성 요소 생성; 프록시 객체로 생성됨
• BatchConfigurerConfiguration -> BasicBatchConfigurer

1. BatchAutoConfiguration
2. BatchProperties 주입받아서 설정 읽고(특정 job만 실행 가능)
3. JobLauncherApplicationRunner 빈 생성
4. JobLauncherApplicationRunner.run 함수 실행할 때 ApplicationArguments(부트 run config에 설정한 값) 받아서 넘겨줌
5. JobLauncherApplicationRunner.execute 함수 실행

<배치 시작>

1. configurer 선언
2. JobBuilderFactory
3. StepBuilderFactory
4. Job; 여러개의 스텝을 하나로
5. Step; 하나의 일
6. tasklet; 작업내용, 비즈니스 로직
7. Job 구동 -> Step 실행 -> tasklet 실행

<메타데이터 저장>

배치가 실행하고 과정을 저장하기 위한 테이블이 필요하다. job에 관련한 테이블 4개, step에 관련한 테이블 2개가 필수적으로 필요하다.(디비 특성에 따라 시퀀스를 저장하는 테이블이 추가될 수 있음)

따라서 DB연결이 필수!