SpringSpring BatchStep

Spring Batch - 10. StepContribution

출처

목차

StepContribution

Chunk 프로세스의 변경 사항을 버퍼링 한 후 StepExecution 상태를 업데이트 하는 객체

  • Chunk Commit 직전에 StepExecution 의 apply 메서드를 호출해 상태르르 업데이트 한다.
  • ExitStatus 의 기본 종료 코드 외 사용자 정의 종료 코드를 생성해서 적용할 수 있다.

  1. TaskletStep 클래스내 doInTransaction 메서드에서 StepExecution 객체를 이용해 StepContribution 객체를 생성한다.
  2. Tasklet 을 실행시킨다.

TaskletStep.java

@Override
public RepeatStatus doInTransaction(TransactionStatus status) {
    TransactionSynchronizationManager.registerSynchronization(this);

    RepeatStatus result = RepeatStatus.CONTINUABLE;

    // StepContribution 객체를 생성한다.
    StepContribution contribution = stepExecution.createStepContribution();

    chunkListener.beforeChunk(chunkContext);
    oldVersion = new StepExecution(stepExecution.getStepName(), stepExecution.getJobExecution());
    copy(stepExecution, oldVersion);

    try {
        try {
            try {
                // Tasklet 을 실행시킨다.
                result = tasklet.execute(contribution, chunkContext);
                if (result == null) {
                    result = RepeatStatus.FINISHED;
                }
            }
            catch (Exception e) {
                if (transactionAttribute.rollbackOn(e)) {
                    chunkContext.setAttribute(ChunkListener.ROLLBACK_EXCEPTION_KEY, e);
                    throw e;
                }
            }
        }
        finally {
            try {
                semaphore.acquire();
                locked = true;
            }
            catch (InterruptedException e) {
                logger.error("Thread interrupted while locking for repository update");
                stepExecution.setStatus(BatchStatus.STOPPED);
                stepExecution.setTerminateOnly();
                Thread.currentThread().interrupt();
            }

            if (logger.isDebugEnabled()) {
                logger.debug("Applying contribution: " + contribution);
            }
            // StepContribution 결과를 StepExecution 객체내에 저장한다.   
            stepExecution.apply(contribution);
        }

        stepExecutionUpdated = true;

        stream.update(stepExecution.getExecutionContext());

        try {
            // Going to attempt a commit. If it fails this flag will
            // stay false and we can use that later.
            getJobRepository().updateExecutionContext(stepExecution);
            stepExecution.incrementCommitCount();
            if (logger.isDebugEnabled()) {
                logger.debug("Saving step execution before commit: " + stepExecution);
            }
            getJobRepository().update(stepExecution);
        }
        catch (Exception e) {
            // If we get to here there was a problem saving the step
            // execution and we have to fail.
            String msg = "JobRepository failure forcing rollback";
            logger.error(msg, e);
            throw new FatalStepExecutionException(msg, e);
        }
    }
    catch (Error e) {
        if (logger.isDebugEnabled()) {
            logger.debug("Rollback for Error: " + e.getClass().getName() + ": " + e.getMessage());
        }
        rollback(stepExecution);
        throw e;
    }
    catch (RuntimeException e) {
        if (logger.isDebugEnabled()) {
            logger.debug("Rollback for RuntimeException: " + e.getClass().getName() + ": " + e.getMessage());
        }
        rollback(stepExecution);
        throw e;
    }
    catch (Exception e) {
        if (logger.isDebugEnabled()) {
            logger.debug("Rollback for Exception: " + e.getClass().getName() + ": " + e.getMessage());
        }
        rollback(stepExecution);
        // Allow checked exceptions
        throw new UncheckedTransactionException(e);
    }
    return result;
}
Share