섹션 11. 스프링 배치 반복 및 오류 제어

[0JUUU]

• [x] Repeat
• [x] FaultTolerant
• [x] Skip
• [x] Retry (1)
• [x] Retry (2)
• [x] Retry (3)
• [x] Skip & Retry 아키텍처

[0JUUU]

Repeat

• 특정 조건이 충족될 때까지 Job 또는 Step을 반복하도록 배치 애플리케이션 구성
• Step과 Chunk의 반복을 RepeatOperation(인터페이스)을 사용해 처리
• 기본 구현체로 RepeatTemplate 제공

1. Step : RepeatTemplate ➡️ Tasklet 반복적으로 실행
2. ChunkOrientedTasklet : 내부적으로 ChunkProvider를 통해 ItemReader한테 데이터를 읽어올 것을 지시
3. ChunkProvider : 내부적으로 RepeatTemplate 가짐 ➡️ 이를 이용해 반복적으로 ItemReader에게 반복적으로 데이터를 읽어오도록 처리

반복 종료 여부를 결정하는 3가지 항목 (3가지 중 하나에 의해 종료됨)

• RepeatStatus
  • 스프링 배치 처리가 끝났는지 판별하기 위한 Enum
    • CONTINUABLE : 작업이 남음
    • FINISHED : 더 이상 반복 ❌
• CompletionPolicy
  • RepeatTemplate의 iterate 메서드 안에서 반복을 중단할지 결정
  • 실행 횟수 또는 완료시기, 오류 발생 시 수행할 작업에 대한 반복여부 결정
  • 정상 종료를 알리는데 사용
  • SimpleCompletePolicy : 기본값으로 현재 반복 횟수가 Chunk 개수보다 크면 반복 종료
  • TimeoutTerminationPolicy : 반복 시점부터 현재 시점까지 소요된 시간이 설정된 시간보다 크면 반복 종료
  • CountingCompletionPolicy : 일정한 카운트를 계산 및 집계 ➡️ 카운트 제한 조건이 만족하면 반복 종료
• ExceptionHandler
  • RepeatCallback 안에서 예외가 발생하면 RepeatTemplate가 ExceptionHandler를 참조해서 예외를 다시 던질지 여부 결정
  • 예외를 받아서 다시 던지게 되면 반복 종료
  • 비정상 종료를 알리는데 사용
  • SimpleLimitExceptionHandler : 기본값으로 예외 타입 중 하나가 발견되면 카운터 증가 & 한계 (limit) 초과되었는지 여부를 확인하고 Throwable을 다시 던짐
  • LogOrRethrowExceptionHandler : 예외를 로그로 기록할지 아니면 예외로 던질지 결정
  • RethrowOnThresholdExceptionHandler : 지정된 유형의 예외가 임계값에 도달하면 다시 발생

@Bean
    public Step step1() throws Exception {
        return stepBuilderFactory.get("step1")
            .<String, String>chunk(5)
            .reader(new ItemReader<String>() {
                int i = 0;

                @Override
                public String read()
                    throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                    i++;
                    return i > 3 ? null : "item" + i;
                }
            })
            .processor(new ItemProcessor<String, String>() {

                RepeatTemplate repeatTemplate = new RepeatTemplate();

                @Override
                public String process(String item) throws Exception {
                    repeatTemplate.setExceptionHandler(simpleLimitExceptionHandler());  // 몇 번의 예외가 발생하면 종료
                    repeatTemplate.iterate(new RepeatCallback() {
                        int i = 0;

                        @Override
                        public RepeatStatus doInIteration(RepeatContext context) throws Exception {
                            i++;
                            System.out.println("repeatTemplate testing " + i);
                            throw new RuntimeException(">> Exception is occurred");
//                            return RepeatStatus.CONTINUABLE;
                        }
                    });
                    return item;
                }
            })
            .writer(System.out::println)
            .build();
    }

• 3번의 예외 발생하기 전까지는 정상적으로 처리
• 이후 예외가 발생하면 예외 처리

[0JUUU]

FaultTolerant

• 오류가 발생해도 Step이 즉시 종료되지 않고 Retry 혹은 Skip 기능을 활성화 ➡️ 내결함성 서비스가 가능하도록 함
• 프로그램의 내결함성을 위해 Skip과 Retry 기능 제공
• Skip
  • ItemReader / ItemProcessor / ItemWriter에 적용 가능
• Retry
  • ItemProcessor / ItemWriter에 적용 가능
• FaultTorelant 구조 : Chunk 기반 프로세스 기반 위에 Skip과 Retry 기능이 추가되어 재정의되어 있음

[0JUUU]

Skip

• 데이터를 처리하는 동안 설정된 Exception 발생 ➡️ 해당 데이터 처리를 건너뛰는 기능
• Step의 실패처리 대신 Skip 진행 ➡️ 배치 수행의 빈번한 실패를 줄일 수 ⭕️

• ItemReader
  • item을 한 건씩 읽다가 예외가 발생하면 해당 Item을 skip하고 다음 item을 읽음
• ItemProcessor
  • item을 처리하다가 예외가 발생하면 해당 Chunk의 첫 단계로 돌아가 itemReader로부터 다시 데이터를 받음
  • 이 때, ItemReader에서 실제 데이터를 다시 읽는 것은 아니고 캐시에 저장한 아이템을 다시 사용해 itemProcessor로 다시 보냄
  • ItemProcessor는 다시 아이템들을 받아 실행하게 되는데 도중 이전 실행에서 예외가 발생했던 정보가 내부적으로 남아있기에 해당 차례가 오면 처리하지 않고 넘어감
  • 결론적으로 skip하는 건 맞는데 itemReader와 동작 방식이 다름
• ItemWriter
  • e.g. Item4에서 예외 발생 ➡️ 다시 Chunk 단위로 ItemReader로 돌아감
  • 캐싱된 데이터로 ItemReader는 ItemProcessor로 넘김
  • ItemProcessor는 하나씩 다시 처리하고 List로 ItemWriter로 보내지 않고 개별로 한 개씩 ItemWriter로 보냄

• 총 2번 예외 발생
  • itemReader에서 i가 3일 때
  • itemProcessor에서 6일 때
• itemReader, itemProcessor, itermWriter 각각에서 횟수를 세는 것이 아닌 모두 통합해서 센다.

  NoSkip 설정

  @Bean
  public Step step1() throws Exception {
      return stepBuilderFactory.get("step1")
          .<String, String>chunk(5)
          .reader(new ItemReader<String>() {
              int i = 0;
              @Override
              public String read() throws NoSkippableException {
                  i++;
                  if(i == 3) {
                      throw new NoSkippableException("skip");
                  }
                  System.out.println("ItemReader : " + i);
                  return i > 20 ? null : String.valueOf(i);
              }
          })
          .processor(itemProcessor())
          .writer(itemWriter())
          .faultTolerant()
          .noSkip(NoSkippableException.class)
          .build();
  }

  

• noSkip에 해당 Exception을 등록 ➡️ skip하지 않고 바로 예외 발생

[0JUUU]

Retry (1)

• ItemProcess, ItemWriter에서 설정된 Exception이 발생 ➡️ 지정한 정책에 따라 데이터 처리를 재시도
• 배치 수행의 빈번한 실패를 줄일 수 ⭕️
• Retry Count는 Item마다 각각 가짐
• 예외 발생 시 재시도 설정에 의해 해당 Chunk의 처음부터 다시 시작
• RetryLimit 횟수 이후에도 재시도가 실패한다면, recover에서 후속작업 처리 가능

• 기존 방식 : ItemProcessor & ItemWriter ➡️ ChunkProvider에서 실행
• Retry 기능 활성화 : ItemProcessor & ItemWriter ➡️ RetryTemplate 안에서 실행
  • 예외 발생 ➡️ RetryTemplate 안에서 처리가 진행
    • ItemProcessor : 다시 Chunk 단계의 처음부터 시작
    • ItemReader는 캐시에 저장된 값은 ItemProcessor로 넘기고 ItemProcessor가 수행되게 함
    • ItemWriter : skip과 다르게 원래대로 List로 한 번에 처리

[0JUUU]

Retry (2)

public class RetryItemProcessor implements ItemProcessor<String, String> {

    private int cnt = 0;

    @Override
    public String process(String item) throws Exception {
        if (item.equals("2") || item.equals("3")) {
            cnt++;
            throw new RetryableException("failed cnt : " + cnt);
        }
        return item;
    }
}

• 2, 3 지나가고 잘 동작할 줄 알았는데 Exception 발생

• 🤔 왜?

  System.out.println("item = " + item);

• 출력해서 확인해보자

  

• 재시도가 일어나면 Chunk의 처음 단위로 넘어감 ➡️ 아예 처음부터 다시 시작해서 0번째부터 restart

• 0 ➡️ 1 ➡️ 2 🔥(예외) ➡️ 0 ➡️ 1 ➡️ 2 🔥 ➡️ 0 ➡️ 1 ➡️ 터져버림💥

• 재시도는 항상 처음부터 시작

• 해결 : 0, 1, 2 예외터지고 재시도할 때 2 실행 안하고 바로 3으로 갈 수 있도록 (예외가 발생했던 부분은 건너뛰고 바로 다음 단계)

• 이걸 Skip으로 해결

• Retry와 Skip으로 함께 해결하면 됨

[0JUUU]

Retry (3)

RetryTemplate 사용

• 기본적으로 제공하는 API가 아닌 RetryTemplate을 사용하여 확장

public class RetryItemProcessor2 implements ItemProcessor<String, Customer> {

    @Autowired
    private RetryTemplate retryTemplate;
    private int cnt = 0;

    @Override
    public Customer process(String item) throws Exception {
        Classifier<Throwable, Boolean> rollbackClassifier = new BinaryExceptionClassifier(true);

        Customer customer = retryTemplate.execute(new RetryCallback<Customer, RuntimeException>() {
            @Override
            public Customer doWithRetry(RetryContext context) throws RuntimeException {
                if (item.equals("1") || item.equals("2")) {
                    cnt++;
                    throw new RetryableException("failed cnt : " + cnt);
                }
                return new Customer(item);
            }
        }, new RecoveryCallback<Customer>() {
            @Override
            public Customer recover(RetryContext context) throws Exception {
                return new Customer(item);
            }
        }, new DefaultRetryState(item, rollbackClassifier));
        return customer;
    }
}

• Recovery에 예외 발생 ➡️ Chunk 단위 처음으로 돌아감
  • 0부터 다시 시작

[0JUUU]

Skip & Retry 아키텍처

ItemReader

ItemProcessor

ItemWriter