Spring Batch job with FlowBuilder fails even though steps complete

Question

I've got a spring batch job that completes the steps successfully but marks the overall batch job status as Failed. The job has two steps. The first step validates the input file and sets a variable for what file type we are processing. That happens with a Tasklet and a JobExecutionDecider. Then a flow builder decides when step to run for step2. Everything runs fine, with each step completing, but the overall batch status is being set to FAILED instead of COMPLETED.

If I check the status of the steps in table BATCH_STEP_EXECUTION (SELECT * FROM BATCH_STEP_EXECUTION WHERE JOB_EXECUTION_ID = 54;) I see that both steps are marked COMPLETED.

Any idea why my overall batch job is being deemed FAILED?

This is what is being returned for the batch status:

2018-07-26 15:21:38.132 INFO 95001 --- [ask-scheduler-1] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=importEmployees]] completed with the following parameters: [{input.file.name=/Users/aaronradich/Dev/posthire-batch/target/classes/data/import/employees.csv, time=1532643668685}] and the following status: [FAILED]

Here's the relevant code. I'm guessing it has something to do with the way the FlowBuilder is configured?:

@Bean
protected Step step1(FlatFileItemReader reader, FileValidationTasklet fileValidationTasklet) {
    return steps.get("step1").tasklet(fileValidationTasklet)
            .build();
}

@Bean
protected Step step2(FlatFileItemReader reader, EmployeeProcessor processor, EmployeeWriter writer) {
    return steps.get("step2")
            .<Employee, Employee>chunk(10)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build();
}

@Bean
public Job importEmployees(@Qualifier("step1") Step step1, @Qualifier("step2") Step step2) {
    JobBuilder jobBuilder = jobBuilderFactory
            .get("importEmployees")
            .incrementer(new RunIdIncrementer());

    StepDecider fileDecider = new StepDecider();

    Flow flowFile = new FlowBuilder<Flow>("fileFlow")
            .start(fileDecider)
                .on(BatchConfig.STATE_DO_FILE_1)
                    .to(step2)
                .on(BatchConfig.STATE_DO_FILE_2)
                    // TODO: Add an additional Step for File Type 2 (maybe called stepF2, etc.)
                    .to(step2)
            .from(fileDecider)
                .on(BatchConfig.STATE_SKIP_FILE)
                    .end(BatchConfig.STATE_FAILED)
                .on("*")
                    .end(BatchConfig.STATE_COMPLETED)
            .build();

    /*
     * Complete Workflow:
     * 
     *           |--> (S2)
     *  S1 ---> -+
     *           |--> (S3)
     */
    FlowJobBuilder builder = jobBuilder
            .flow(step1)
            .next(flowFile)
            .end();

    return builder.build();
}

Here is the JobExecutionDecider:

package com.sterlingts.posthire.batch;

import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;

import com.sterlingts.posthire.batch.config.BatchConfig;

public class StepDecider implements JobExecutionDecider {

    @Override
    public FlowExecutionStatus decide(org.springframework.batch.core.JobExecution jobExecution,
            org.springframework.batch.core.StepExecution stepExecution) {

        String fileTypeCode = jobExecution.getExecutionContext().getString("FileTypeCode");
        String returnStatus = "";

        switch(fileTypeCode) {
            case "1":
                returnStatus = BatchConfig.STATE_DO_FILE_1;
                break;
            case "2":
                returnStatus = BatchConfig.STATE_DO_FILE_2;
                break;
            default:
                returnStatus = BatchConfig.STATE_SKIP_FILE;
                break;
        }

        return new FlowExecutionStatus(returnStatus);

    }
}

And here is the TaskLet that determines if the file is valid and sets a variable in the JobContext for the file type we are processing:

package com.sterlingts.posthire.batch.util;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

public class FileValidationTasklet implements Tasklet {

    private final String FILE_HEADER_1 = "EmployeeId,SSN,FirstName,MiddleName,LastName,Suffix,MaidenName,DOB,Gender,Country,Street,AptNo,City,Region,PostalCode,PhoneDay,PhoneDayExt,PhoneEve,PhoneEveExt,Email,Account,EmployeeStatus,HireDate,StartDate,TerminateDate,DLNumber,DLRegion,LastNameOnLicense,FirstNameOnLicense,MiddleNameOnLicense,JobPosition,BillCode";
    private final String FILE_HEADER_2 = "EmployeeId,SSN,FirstName,MiddleName,LastName";

    private static final Logger log = LoggerFactory.getLogger(FileValidationTasklet.class);

    private String inputFile;
    private String fileTypeCode;

    public FileValidationTasklet() {

    }

    public FileValidationTasklet(String inputFile) {
        this.inputFile = inputFile;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {        
        System.out.println("Input file is " + inputFile);

        if (this.isValidFile()) {
            // store the FileTypeCode in the jobExecutionContext so we can access it in other steps
            chunkContext
                    .getStepContext()
                    .getStepExecution()
                    .getJobExecution()
                    .getExecutionContext()
                    .put("FileTypeCode", this.fileTypeCode);
        } else {
            throw new Exception("File '" + inputFile + "' is not a valid file format!  Aborting import!");
        } 

        // set the exit status for the Tasklet to Completed so the overall job can get completed
        //chunkContext.getStepContext().getStepExecution().setExitStatus(ExitStatus.COMPLETED);
        chunkContext.getStepContext().getStepExecution().setStatus(BatchStatus.COMPLETED);

        // tell Batch to continue on to the next step and not repeat this one
        return RepeatStatus.FINISHED;
    }

    private Boolean isValidFile() {
        Boolean validFile = false;

        BufferedReader brFile;
        try {
            brFile = new BufferedReader(new FileReader(inputFile));
            String headerRow = brFile.readLine();

            // have noticed that sometimes the leading character from the readLine is a space
            headerRow = headerRow.trim();

            // strip CR and LF just in case the file originated from Windows and still has a CR or LF after the readLine
            headerRow = headerRow.replaceAll("\r", "").replaceAll("\n", "");

            // remove the leading byte-order mark (BOM) character
            headerRow = SpringIntegrationUtils.removeUTF8BOM(headerRow);

            if (headerRow.equals(FILE_HEADER_1)) {
                this.fileTypeCode = "1";
                validFile = true;
            } else if (headerRow.equals(FILE_HEADER_2)) {
                this.fileTypeCode = "2";
                validFile = true;
            }

        } catch (FileNotFoundException e) {
            log.error("File '" + inputFile + "' was not found!  Aborting!");
        } catch (IOException e) {
            log.error("Error validating header row of file '" + inputFile + "'!");
        }

        return validFile;
    }

}