Spring batch conditional flow creates infinite loop

Question

I have a simple 3-step flow:

public Job myJob() {
    Step extract = extractorStep();
    Step process = filesProcessStep();
    Step cleanup = cleanupStep();

    return jobBuilderFactory.get("my-job")
          .flow(echo("Starting job"))
          .next(extract)
          .next(process)
          .next(cleanup)
          .next(echo("Ending job"))
          .end()
          .build();
  }

Now I want to add error processing using ExitStatus from StepExecutionListener.afterStep(). Any error should forward flow to cleanup step. So I changed to the code below:

 public Job myJob() {
    Step extract = extractorStep();
    Step process = filesProcessStep();
    Step cleanup = cleanupStep();

    return jobBuilderFactory.get("my-job")
          .start(echo("Starting batch job"))

          .next(extract).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
          .from(extract).on("*").to(process)

          .next(process).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
          .from(process).on("*").to(cleanup)

          .next(echo("End batch job"))
          .end()
          .build();
  }

Now I have an infinite loop to the cleanup step. I would like some help to correct this flow.

user_5019386 · Answer

In your example, the flow is undefined starting from cleanup. You should precise that from cleanup the flow must continue to echo using .from(cleanup).to(echo("End batch job")).end(). Here is an example:

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class MyJob {

    @Autowired
    private JobBuilderFactory jobs;

    @Autowired
    private StepBuilderFactory steps;

    @Bean
    public Step extractorStep() {
        return steps.get("extractorStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("extractorStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step filesProcessStep() {
        return steps.get("filesProcessStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("filesProcessStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step cleanupStep() {
        return steps.get("cleanupStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("cleanupStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    public Step echo(String message) {
        return steps.get("echo-" + message)
                .tasklet((contribution, chunkContext) -> {
                    System.out.println(message);
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Job job() {
        Step start = echo("Starting batch job");
        Step extract = extractorStep();
        Step process = filesProcessStep();
        Step cleanup = cleanupStep();
        Step stop = echo("End batch job");

        return jobs.get("job")
                .start(start).on("*").to(extract)

                    .from(extract).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
                    .from(extract).on("*").to(process)

                    .from(process).on("*").to(cleanup)

                    .from(cleanup).next(stop)
                    .build()

                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

}

it prints:

Starting batch job
extractorStep
filesProcessStep
cleanupStep
End batch job

if the extractorStep fails, for example:

@Bean
public Step extractorStep() {
    return steps.get("extractorStep")
            .tasklet((contribution, chunkContext) -> {
                System.out.println("extractorStep");
                chunkContext.getStepContext().getStepExecution().setExitStatus(ExitStatus.FAILED);
                return RepeatStatus.FINISHED;
            })
            .build();
}

the flow will skip filesProcessStep and go to cleanup:

Starting batch job
extractorStep
cleanupStep
End batch job

Hope this helps.

Spring batch conditional flow creates infinite loop

1 Answers