Spring Batch 입문 41편 — Common Patterns · 흔한 운영 패턴 카탈로그

이 글은 Spring Batch 입문에서 운영까지 시리즈 48편 중 41편이에요. 40편 까지 Part 9 (Repeat·Retry·Testing) 를 마쳤다면, 이번 41편부터 Part 10 — Patterns · Integration · Observability 로 들어갑니다. 첫 글은 공식 문서가 정리한 흔한 패턴 카탈로그예요.

8가지 패턴 — 한눈에

1. Logging Item Failures

18편 ItemReadListener·ItemWriteListener (read·write 시점 콜백) 의 정확한 응용이에요.

public class ItemFailureLoggerListener extends ItemListenerSupport {

    private static final Logger log = LoggerFactory.getLogger("item.error");

    @Override
    public void onReadError(Exception ex) {
        log.error("Encountered error on read", ex);
    }

    @Override
    public void onWriteError(Exception ex, List<? extends Object> items) {
        log.error("Encountered error on write for {} items", items.size(), ex);
    }
}

logger name 을 따로 둔 덕에 (item.error) 로그 채널이 분리됩니다.

등록

@Bean
public Step step(JobRepository repo, PlatformTransactionManager tx) {
    return new StepBuilder("step", repo)
        .<X, Y>chunk(100, tx)
        .reader(...)
        .writer(...)
        .listener(new ItemFailureLoggerListener())
        .build();
}

함정 — onError 에서 transaction

If your listener does anything in an onError() method, it must be inside a transaction that is going to be rolled back. — 공식 reference

onError 안에서 DB 호출을 하면 같은 transaction 이 rollback 되면서 로그도 같이 사라집니다.

해결 — @Transactional(propagation = REQUIRES_NEW) 로 별도 transaction 을 떼어내세요.

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void logErrorToDatabase(Exception ex) {
    errorRepo.save(new ErrorLog(ex.getMessage()));
}

또는 18편 SkipListener (commit 직전 호출이 보장되는 listener) 를 쓰면 더 안전해요.

2. Stopping Job Manually — 3가지 방법

방법 A: 예외 throw

public class PoisonPillItemProcessor<T> implements ItemProcessor<T, T> {
    @Override
    public T process(T item) {
        if (isPoisonPill(item)) {
            throw new PoisonPillException("Poison pill: " + item);
        }
        return item;
    }
}

14편 skip / 15편 retry 가 흡수해버릴 수 있으니, retry·skip 대상에서 빠진 예외를 쓰거나 retry·skip 자체를 끄세요.

방법 B: Reader 가 null 반환

public class EarlyCompletionItemReader<T> implements ItemReader<T> {
    private ItemReader<T> delegate;

    @Override
    public T read() throws Exception {
        T item = delegate.read();
        if (isEndItem(item)) {
            return null;     // ★ 조기 종료 신호
        }
        return item;
    }
}

22편의 null = 종료 contract 를 활용한 방식입니다. 자연스럽게 끝나니까 BatchStatus 는 COMPLETED 가 돼요.

방법 C: StepExecution.setTerminateOnly()

public class TerminatingListener extends ItemListenerSupport
        implements StepListener {

    private StepExecution stepExecution;

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public void afterRead(Object item) {
        if (isPoisonPill(item)) {
            stepExecution.setTerminateOnly();    // ★ flag 설정
        }
    }
}

setTerminateOnly() 는 종료 flag (중단 요청 표식) 만 세웁니다. 다음 item 처리 직전에 프레임워크가 이 flag 를 확인하고 JobInterruptedException 을 던지면서 BatchStatus.STOPPED 로 마무리해요.

3 방법 비교

20편 Flow Control 의 end()·fail()·stopAndRestart() 와 그대로 짝이 맞아떨어집니다.

3. Footer with Summary

30편 headerCallback·footerCallback 의 정확한 응용에 집계 상태 추적이 더 붙은 패턴이에요.

public class TradeItemWriter implements ItemWriter<Trade>, FlatFileFooterCallback,
                                         ItemStream {

    private ItemWriter<Trade> delegate;
    private BigDecimal totalAmount = BigDecimal.ZERO;

    @Override
    public void write(Chunk<? extends Trade> items) throws Exception {
        BigDecimal chunkTotal = BigDecimal.ZERO;
        for (Trade trade : items) {
            chunkTotal = chunkTotal.add(trade.getAmount());
        }

        delegate.write(items);

        totalAmount = totalAmount.add(chunkTotal);   // write 성공 후 합산
    }

    @Override
    public void writeFooter(Writer writer) throws IOException {
        writer.write("Total Amount Processed: " + totalAmount);
    }

    // ★ Restart 안전성
    @Override
    public void open(ExecutionContext context) {
        if (context.containsKey("total.amount")) {
            totalAmount = (BigDecimal) context.get("total.amount");
        }
    }

    @Override
    public void update(ExecutionContext context) {
        context.put("total.amount", totalAmount);
    }

    @Override
    public void close() {}
}

여기서 봐야 할 게 세 가지예요. 첫째, write 가 성공한 뒤에야 합산을 더해서 rollback 에 안전합니다. 둘째, ItemStream (Step 상태를 저장·복구하는 인터페이스) 을 구현했으니 restart 때 합산 값이 그대로 살아나요. 셋째, FlatFileFooterCallback (파일 끝에 라인을 쓰는 콜백) 을 delegate Writer 의 footerCallback 으로 등록해줘야 실제로 호출됩니다.

등록

@Bean
public FlatFileItemWriter<Trade> delegateWriter(TradeItemWriter aggregator) {
    return new FlatFileItemWriterBuilder<Trade>()
        .name("delegateWriter")
        .resource(...)
        .delimited().delimiter(",").names("id", "amount")
        .footerCallback(aggregator)      // ★ aggregator 가 footerCallback
        .build();
}

4. Driving Query Pattern

큰 테이블을 cursor 로 훑으면 pessimistic lock (행 잠금) 이 길어지고 online 시스템까지 영향을 받습니다. 그래서 key 만 select 한 다음, 각 key 별로 따로 fetch 하는 식으로 풉니다.

@Bean
public JdbcCursorItemReader<Long> keyReader(DataSource ds) {
    return new JdbcCursorItemReaderBuilder<Long>()
        .name("keyReader")
        .dataSource(ds)
        .sql("SELECT id FROM foo WHERE status = 'NEW'")
        .rowMapper((rs, rowNum) -> rs.getLong("id"))    // ID 만
        .build();
}

@Bean
public ItemProcessor<Long, Foo> fooEnrichProcessor(FooDao dao) {
    return id -> dao.findById(id);      // ID 로 detail fetch
}

@Bean
public Step drivingStep(JobRepository repo, PlatformTransactionManager tx,
                        JdbcCursorItemReader<Long> reader,
                        ItemProcessor<Long, Foo> processor,
                        ItemWriter<Foo> writer) {
    return new StepBuilder("drivingStep", repo)
        .<Long, Foo>chunk(100, tx)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .build();
}

장점은 분명해요. cursor 가 들고 있는 결과가 ID 만이라 lock 부담이 줄고, 각 fetch 도 짧은 query 라 online 시스템에 덜 부담을 줍니다. 36편 reusing services 처럼 기존 DAO 를 그대로 재사용할 수 있다는 것도 큽니다.

단점은 N+1 호출. chunk 100 이면 1 + 100 = 101 번 query 가 나가서 대량 환경에서는 throughput 이 떨어집니다. 결국 DB 부하 분산과 throughput 사이의 tradeoff 라, online 시스템 영향이 큰 환경에서 쓰는 패턴이에요.

5. Multi-Line Records

입력 예시:

HEA;0013100345;2007-02-15
NCU;Smith;Peter;;T;20014539;F
BAD;;Oak Street 31/A;;Small Town;00235;IL;US
FOT;2;2;267.34

HEA 부터 FOT 까지가 한 record. 즉 4 line 이 한 개의 Trade 객체로 묶여요.

Wrapping Reader

public class MultiLineTradeItemReader implements ItemReader<Trade>, ItemStream {

    private FlatFileItemReader<FieldSet> delegate;

    @Override
    public Trade read() throws Exception {
        Trade t = null;

        for (FieldSet line = null; (line = delegate.read()) != null;) {
            String prefix = line.readString(0);

            if (prefix.equals("HEA")) {
                t = new Trade();    // record 시작
            } else if (prefix.equals("NCU")) {
                Assert.notNull(t, "No header found");
                t.setLast(line.readString(1));
                t.setFirst(line.readString(2));
                // ...
            } else if (prefix.equals("BAD")) {
                t.setCity(line.readString(4));
                t.setState(line.readString(6));
            } else if (prefix.equals("FOT")) {
                return t;           // record 끝 → 반환
            }
        }
        Assert.isNull(t, "No 'FOT' found");
        return null;
    }

    // ItemStream 위임 (17편 Delegate Pattern)
    @Override
    public void open(ExecutionContext c) { delegate.open(c); }
    @Override
    public void update(ExecutionContext c) { delegate.update(c); }
    @Override
    public void close() { delegate.close(); }
}

Delegate Reader

@Bean
public FlatFileItemReader<FieldSet> flatFileItemReader() {
    return new FlatFileItemReaderBuilder<FieldSet>()
        .name("multiLineDelegate")
        .resource(new ClassPathResource("data/multiLine.txt"))
        .lineTokenizer(orderFileTokenizer())
        .fieldSetMapper(new PassThroughFieldSetMapper())   // ★ FieldSet 그대로
        .build();
}

@Bean
public PatternMatchingCompositeLineTokenizer orderFileTokenizer() {
    Map<String, LineTokenizer> tokenizers = new HashMap<>();
    tokenizers.put("HEA*", headerRecordTokenizer());
    tokenizers.put("FOT*", footerRecordTokenizer());
    tokenizers.put("NCU*", customerLineTokenizer());
    tokenizers.put("BAD*", billingAddressLineTokenizer());

    PatternMatchingCompositeLineTokenizer tokenizer =
        new PatternMatchingCompositeLineTokenizer();
    tokenizer.setTokenizers(tokenizers);
    return tokenizer;
}

작동 방식을 풀어보면, delegate 가 line 단위로 FieldSet 을 반환하고 PassThroughFieldSetMapper (FieldSet 을 변환 없이 그대로 넘기는 mapper) 가 그걸 그대로 통과시켜요. 그러면 wrapper 가 prefix 를 검사해서 한 record 로 조립합니다. PatternMatchingCompositeLineTokenizer (prefix 패턴별로 다른 tokenizer 를 라우팅) 가 prefix 별로 다른 tokenizer 를 골라주는 역할이에요 (29편).

6. Executing System Commands

19편 SystemCommandTasklet (외부 OS 명령을 Step 안에서 실행) 의 정확한 응용입니다.

@Bean
public SystemCommandTasklet tasklet() {
    SystemCommandTasklet tasklet = new SystemCommandTasklet();
    tasklet.setCommand("echo hello");
    tasklet.setTimeout(5000);
    return tasklet;
}

batch 안에서 굳이 외부 command 를 돌리는 이유는, JobRepository 에 Job metadata 가 통합되고, multi-step job 의 한 단계로 묶이고, 재시작도 안전하게 처리되기 때문이에요. 별도 scheduler 로 분리하면 metadata 가 떨어져나가고 재시작도 복잡해집니다.

7. Handling No Input as Failure

public class NoWorkFoundStepExecutionListener implements StepExecutionListener {
    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        if (stepExecution.getReadCount() == 0) {
            return ExitStatus.FAILED;
        }
        return null;        // 다른 변경 없음 → 원래 status 유지
    }
}

Spring Batch 의 기본 동작은 0건을 처리해도 COMPLETED 예요 (자연스러운 흐름). 그런데 파일 이름이 misnamed 라 0건이 read 됐는데 COMPLETED 가 떨어지면 의도와 어긋나고, 0건 자체가 예외 신호인 경우에는 명시적으로 fail 처리해줘야 합니다.

그래서 Listener 가 afterStep 에서 readCount 를 검사하고, 0이면 FAILED 를 반환하게 둬요.

등록

@Bean
public Step step(JobRepository repo, PlatformTransactionManager tx) {
    return new StepBuilder("step", repo)
        .<X, Y>chunk(100, tx)
        .reader(...)
        .writer(...)
        .listener(new NoWorkFoundStepExecutionListener())
        .build();
}

8. Passing Data Between Steps — Promotion Listener

문제

Step1 에서 모은 통계를 Step2 가 쓰고 싶거나, Step1 의 처리 시각을 Step2 가 기준으로 삼고 싶은 상황을 떠올려보세요.

ExecutionContext (Step·Job 단위 상태 저장소) 는 두 가지가 있어요.

Step 끼리 데이터를 공유하려면 Job ExecutionContext 를 써야 하는데, Step 안에서 직접 Job ExecutionContext.put 을 호출하면 Step 이 실패했을 때 데이터가 같이 날아갑니다 (Step 이 끝나야 update 되니까요). 그래서 안전한 방법은 Step ExecutionContext 에 일단 저장하고, Step 이 끝나면 Job ExecutionContext 로 promote 시키는 거예요.

SavingItemWriter — Step ExecutionContext 에 저장

public class SavingItemWriter implements ItemWriter<Object> {
    private StepExecution stepExecution;

    @BeforeStep
    public void saveStepExecution(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public void write(Chunk<? extends Object> items) throws Exception {
        // 처리
        ExecutionContext stepContext = stepExecution.getExecutionContext();
        stepContext.put("someKey", computedValue);
    }
}

ExecutionContextPromotionListener

@Bean
public ExecutionContextPromotionListener promotionListener() {
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    listener.setKeys(new String[]{"someKey"});
    // optional: listener.setStatuses(new String[]{"COMPLETED"});
    return listener;
}

@Bean
public Step step1(JobRepository repo, PlatformTransactionManager tx,
                  ItemWriter<X> savingWriter,
                  ExecutionContextPromotionListener promotionListener) {
    return new StepBuilder("step1", repo)
        .<X, X>chunk(10, tx)
        .reader(reader())
        .writer(savingWriter)
        .listener(promotionListener)
        .build();
}

ExecutionContextPromotionListener (Step→Job context 로 key 를 복사해주는 listener) 는 Step 이 끝나고 ExitStatus 가 매칭되면 (default 가 COMPLETED) 지정 key 들을 Step ExecutionContext 에서 Job ExecutionContext 로 그대로 copy 해줘요.

RetrievingItemWriter — Step2 에서 읽기

public class RetrievingItemWriter implements ItemWriter<Object> {
    private Object someObject;

    @BeforeStep
    public void retrieveInterstepData(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        ExecutionContext jobContext = jobExecution.getExecutionContext();
        this.someObject = jobContext.get("someKey");
    }

    @Override
    public void write(Chunk<? extends Object> items) throws Exception {
        // someObject 활용
    }
}

21편 Late Binding 결합

Step2 의 reader/writer 를 @StepScope + @Value("#{jobExecutionContext['someKey']}") 로 받으면 더 깔끔해져요 (24편의 Step 간 데이터 전달 패턴).

자주 만나는 사고

사고 1: Logging Listener 의 transaction 충돌

원인 — onError 안에서 호출한 DB 작업이 현재 transaction 안에 묶여서, rollback 될 때 log 도 같이 사라지는 거예요.

해결 — @Transactional(REQUIRES_NEW) 또는 SkipListener (commit 직전 보증).

사고 2: setTerminateOnly 가 즉시 멈추지 않음

원인 — setTerminateOnly() 는 flag 만 세우니까 다음 item 처리 전까지는 현재 chunk 가 그대로 굴러갑니다.

해결 — 즉시 중단이 필요하면 예외 throw.

사고 3: Footer summary 의 restart 문제

원인 — totalAmount 가 내부 state 인데 ItemStream 을 안 구현해놨을 때 생겨요.

해결 — ItemStream 구현 + ExecutionContext put/get.

사고 4: Footer summary 의 chunk rollback

원인 — totalAmount 를 write 전에 합산해버리면 rollback 이 나도 합산이 그대로 남아 계산이 어긋납니다.

해결 — write 성공 후 합산 (위 예제 참고).

사고 5: Driving Query 의 N+1

원인 — 각 ID 별로 fetch 가 나가니 query 수가 폭증해요.

해결 — batch fetch (WHERE id IN (?, ?, ?, ...)) 또는 chunk 단위 Processor.

사고 6: Multi-Line Records 의 header 없음

원인 — 파일이 손상돼서 HEA 가 빠진 record 가 들어올 때입니다.

해결 — Assert 또는 명시 예외 → skip 흡수.

사고 7: Promotion Listener 가 안 작동

원인 1 — setKeys 를 지정하지 않은 경우. 원인 2 — Step ExitStatus 가 listener 의 setStatuses 와 어긋난 경우. 원인 3 — Step ExecutionContext 에 해당 key 자체가 없는 경우.

해결 — keys · statuses · Step ExecutionContext 모두 점검.

운영 권장 패턴 요약

Logging 채널 분리

@Bean
public ItemFailureLoggerListener errorLogger() {
    return new ItemFailureLoggerListener();
}

item.error logger name 을 logback config 에서 별도 file appender 로 빼주세요.

비즈니스 중단

대부분 — setTerminateOnly() (STOPPED, 운영자 검토 후 재시작) 즉시 — 예외 throw (FAILED) 자연 — Reader null (COMPLETED)

Footer summary

ItemStream 구현 + write 성공 후 합산 조합이 안전합니다.

Driving Query

Online 시스템 영향 minimize 필요 = key cursor + DAO fetch. Throughput 우선 = full cursor + RowMapper.

Multi-Line

Wrapping Reader + PatternMatchingComposite + PassThroughFieldSetMapper 의 표준 trio.

Step 간 데이터

ExecutionContextPromotionListener 가 공식 패턴이고, 21편 Late Binding 으로 깔끔하게 받습니다.

시험 직전 한 번 더 — Common Patterns 함정 압축 노트

• 8 패턴 = Logging Failures · Stopping · Footer Summary · Driving Query · Multi-Line · SystemCommand · NoWorkFound · Step 간 데이터
• Logging = ItemListenerSupport 의 onReadError · onWriteError (별도 logger name)
• onError 안 DB 호출 = REQUIRES_NEW transaction 또는 SkipListener (commit 직전 보증)
• Stopping 3 방법 = 예외 (FAILED) · null (COMPLETED) · setTerminateOnly() (STOPPED)
• setTerminateOnly() = flag 설정, 다음 item processing 전 검사
• Footer Summary = FlatFileFooterCallback + ItemWriter 동시 구현
• write 성공 후 합산 (rollback 안전)
• ItemStream 구현 필수 (재시작 시 합산 값 복구)
• open() 에서 ExecutionContext 복구 · update() 에서 저장
• Driving Query = key 만 select 후 ItemProcessor 에서 detail fetch
• DB cursor lock 부담 회피, online 시스템 영향 최소화
• 단점 — N+1 호출, throughput ↓
• Multi-Line = Wrapping Reader + PatternMatchingCompositeLineTokenizer + PassThroughFieldSetMapper
• delegate 가 FieldSet 단위 → wrapper 가 prefix 검사로 record 조립
• record 종료 = FOT prefix 만나면 return
• SystemCommandTasklet = 외부 OS 명령 + Job metadata 통합 + timeout
• NoWorkFoundStepExecutionListener = readCount == 0 → ExitStatus.FAILED
• 기본 Spring Batch = 0건 처리도 COMPLETED
• file misnamed 같은 실수 감지 자리
• Step 간 데이터 전달 — Job ExecutionContext 가 Step 종료 시 update
• Step 안 직접 Job context put = Step 실패 시 손실
• 안전 = Step ExecutionContext put → ExecutionContextPromotionListener 가 promote
• Promotion Listener = setKeys + (옵션) setStatuses (default = COMPLETED)
• 받기 = Job ExecutionContext.get 또는 21편 #{jobExecutionContext['key']} Late Binding
• 함정 — onError transaction 충돌 (REQUIRES_NEW 또는 SkipListener)
• 함정 — setTerminateOnly 즉시 X (다음 item 전 검사)
• 함정 — Footer ItemStream 미구현 restart 시 합산 0
• 함정 — Footer chunk rollback (write 성공 후 합산)
• 함정 — Driving Query N+1 (batch fetch 또는 chunk processor)
• 함정 — Multi-Line header 없음 (Assert 또는 명시 예외)
• 함정 — Promotion Listener 안 작동 (keys/statuses/context 점검)
• 패턴 — Logging 채널 분리 (logger name + appender)
• 패턴 — Stopping = setTerminateOnly (운영자 개입), 예외 (즉시), null (자연)
• 패턴 — Footer = ItemStream + write 후 합산
• 패턴 — Driving Query = online 시스템 영향 최소화
• 패턴 — Multi-Line = Wrapping Reader trio
• 패턴 — Step 간 데이터 = Promotion + Late Binding

공식 문서: Common Batch Patterns 에서 원문을 확인할 수 있어요.

시리즈 다른 편 (앞뒤 글 모음)

이전 글:

• 36편 — Reusing Services · ItemReaderAdapter · Process Indicator
• 37편 — Scaling · Parallel 6가지 전략 종합
• 38편 — Repeat · RepeatTemplate · CompletionPolicy
• 39편 — Retry · Spring Framework 7 Core Retry (v6 변경)
• 40편 — Testing · @SpringBatchTest · End-to-End

다음 글:

• 42편 — Spring Batch Integration · 두 프레임워크 경계
• 43편 — Launching via Messages · JobLaunchingGateway · Informational
• 44편 — Async Processors · Remote Chunking · Partitioning
• 45편 — Observability · Micrometer · Tracing
• 46편 — Java Flight Recorder (JFR) · v6 신규