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.

Answer 1

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.