spring-batch-fault-tolerant

@Slf4j
public class SimpleReader implements ItemReader<String> {

    private ConcurrentLinkedQueue<String> queue;

    /**
     * for spring bean init
     */
    public void init() {
        log.info("init simple reader...");
        queue = new ConcurrentLinkedQueue<>();
        queue.addAll(Arrays.asList(
            "message 1", "message 2", "message 3", "message 4", "message 5",
            "message 6", "message 7", "message 8", "message 9"
        ));
    }

    @Override
    public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        // 一次輸出一個 message, 直到 queue 為空
        return queue.isEmpty() ? null : queue.poll();
    }
}

@Slf4j
public class SimpleProcessor implements ItemProcessor<String, String> {
    
    @Override
    public String process(String msg) throws Exception {
        if ("message 5".equals(msg)) {
            // 簡單示範 message 5, 處理時必定會拋出自定義的 RetryException 
            throw new RetryException("message 5 was FAILED");
        }
        if ("message 8".equals(msg)) {
            // 簡單示範 message 8, 處理時必定會拋出自定義的 RetryException
            throw new RetryException("message 8 was FAILED");
        }
        log.info("processing : {}", msg);
        return msg.toUpperCase();
    }
}

@Slf4j
public class RetryException extends Exception {

    public RetryException(String message) {
        super(message);
        log.warn("Catch Retry Exception: {}", message); 
    }
}

@Slf4j
public class SimpleWriter implements ItemWriter<String> {
    
    @Override
    public void write(List<? extends String> msgList) throws Exception {
        msgList.forEach(System.out::println);
    }
}

@Configuration
@EnableBatchProcessing
public class FooJob {

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;


    @Bean(initMethod = "init")
    @StepScope
    public SimpleReader simpleReader() {
        return new SimpleReader();
    }

    @Bean
    @StepScope
    public SimpleProcessor simpleProcessor() {
        return new SimpleProcessor();
    }

    @Bean
    @StepScope
    public SimpleWriter simpleWriter() {
        return new SimpleWriter();
    }

    @Bean
    public Job simpleJob() {
        return jobBuilderFactory.get("SIMPLE_JOB")
            .incrementer(new RunIdIncrementer())
            .flow(simpleStep())     // 在操作設計上, 個人比較喜歡用 flow, 在設計上更方便組合 step
            .end()
            .build();
    }

    @Bean
    public Step simpleStep() {
        return stepBuilderFactory.get("SIMPLE_STEP")
            .<String, String>chunk(1)
            .reader(simpleReader())
            .processor(simpleProcessor())
            .writer(simpleWriter())
            .faultTolerant()                    // 開啟容錯設定
            .retry(RetryException.class)        // 當這個 step 某個階段拋出 RetryException
            .backOffPolicy(backOfFiveSec())     // wait 5 sec
            .retryLimit(3)                      // 最多 retry 3 次
            .skip(RetryException.class)         // 當這個 step 某個階段拋出 RetryException 做 skip 處理
            .skipLimit(1)                       // 最多 skip 1 次
            .build();
    }

    private FixedBackOffPolicy backOfFiveSec() {
        FixedBackOffPolicy fiveSecBackoff = new FixedBackOffPolicy();
        fiveSecBackoff.setBackOffPeriod(5_000);
        fiveSecBackoff.setSleeper(new ThreadWaitSleeper());
        return fiveSecBackoff;
    }
}

@Slf4j
@Component
@AllArgsConstructor
public class LaunchJob implements InitializingBean {

    private JobLauncher jobLauncher;
    private Job job;

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("launching foo-job...");
        final JobParameters jobParameters = new JobParametersBuilder()
            .addLong("execution", System.currentTimeMillis())
            .toJobParameters();
        jobLauncher.run(job, jobParameters);
    }
}