将 Spring 批处理任务失败消息传递到报告步骤

Question

我正在使用带有OpenCSV的Spring Batch Tasklet来读取我的CSV文件。在提出问题之前，我知道块，但在后面的步骤中文件之间存在交叉验证，因此我必须继续使用 Tasklet。

我正在尝试做的是向报告步骤报告丢失的文件或解析错误。我不确定将失败报告到下一步的正确方法是什么。我有以下代码。

读取文件的初始步骤。

public class CsvBatchReader<T> implements Tasklet, StepExecutionListener {
private final Logger logger = LoggerFactory.getLogger(CsvBatchReader.class);
private List batch;
private final Class<T> clazz;
private Path path;
public CsvBatchReader(Class<T> clazz, Path path) {
this.clazz = clazz;
this.path = path;
}
@Override
public void beforeStep(StepExecution stepExecution) {
logger.info("Reader initialized - " + clazz.getSimpleName());
batch = new ArrayList();
}
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
logger.info("Reader ended - " + clazz.getSimpleName());
return ExitStatus.COMPLETED;
}
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws UnexpectedJobExecutionException {
logger.info("Reader execute - " + clazz.getSimpleName());

ICsvToBean csvToBean = new CsvToBean(clazz, path);
try {
batch = csvToBean.readCsv();
} catch(IOException ex) {
// error message being caught from my csvToBean class. 
throw new UnexpectedJobExecutionException("Invalid file " + ex.getMessage());
}
return RepeatStatus.FINISHED;
}
}

报告步骤

我不确定如何在不使用步骤执行上下文的情况下传入异常消息，或者是否有定义的方法来传入失败消息。

public class CsvBatchReporting implements Tasklet, StepExecutionListener {
private final Logger logger = LoggerFactory.getLogger(CsvBatchCrossValidation.class);
private List errorMessages;
private List skippedInserts;
@Override
public void beforeStep(StepExecution stepExecution) {
logger.info("Reporting initialized");
ExecutionContext executionContext = stepExecution
.getJobExecution()
.getExecutionContext();
System.out.println("description " + stepExecution.getStatus());

}
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
logger.info("Reporting ended");
return ExitStatus.COMPLETED;
}
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
logger.info("Reporting execute");
//Email Error 
return RepeatStatus.FINISHED;
}
}

作业配置

@Bean
public Job primaryCareJob(@Qualifier("reportingStep") Step reportingStep, @Qualifier("crossValidationStep") Step crossValidationStep) {
logger.info("Start PrimaryCare Job");
return jobs.get("primaryCareJob")
.start(readPrimaryCareStep()).on("FAILED").to(reportingStep)
.from(readPrimaryCareStep()).on("*").to(readPrimaryCareDetailStep())
.from(readPrimaryCareDetailStep()).on("FAILED").to(reportingStep)
.from(readPrimaryCareDetailStep()).on("*").to(processPrimaryCareStep())
.from(processPrimaryCareStep()).on("INVALID").to(reportingStep)
.from(processPrimaryCareStep()).on("*").to(processPrimaryCareDetailStep())
.from(processPrimaryCareDetailStep()).on("INVALID").to(reportingStep)
//Other steps
.from(reportingStep).on("*").end()
.from(reportingStep).on("*").fail()
.build()
.build();
}

我开始将作业模式更改为失败，而不是将其定义为无效以使异常自动调用失败的步骤。我使用无效的其他步骤正在我的afterStep中使用以下代码定义。

if(!errorMessages.isEmpty()) {
chunkContext.getStepContext().getStepExecution().setExitStatus(new ExitStatus("INVALID"));
}

如何从阅读器获取 CSV 异常消息以传递到我的报告步骤，以便我可以将其作为电子邮件发送？

Answer 1

我不确定如何传入异常消息，或者是否有定义的方法可以在不使用步骤执行上下文的情况下传入失败消息。

您可以从作业执行中访问上一步中引发的异常。下面是一个示例：

import java.util.List;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
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 step1() {
return steps.get("step1")
.tasklet((contribution, chunkContext) -> {
System.out.println("hello");
throw new Exception("Boom!");
})
.build();
}
@Bean
public Step step2() {
return steps.get("step2")
.tasklet((contribution, chunkContext) -> {
JobExecution jobExecution = chunkContext.getStepContext().getStepExecution().getJobExecution();
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); // TODO properly get the stepExecution of the previous step
List<Throwable> failureExceptions = stepExecution.getFailureExceptions();
if (!failureExceptions.isEmpty()) {
Throwable throwable = failureExceptions.get(0);
System.out.println("Looks like step1 has thrown an exception: " + throwable.getMessage());
}
System.out.println("world");
return RepeatStatus.FINISHED;
})
.build();
}
@Bean
public Job job() {
return jobs.get("job")
.flow(step1())
.on("*").to(step2())
.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());
}
}

此示例打印：

hello
Looks like step1 has thrown an exception: Boom!
world

显然，您需要确保步骤1在所有情况下都流向步骤2(因此流定义(。

希望这有帮助。

Answer 2

考虑将errorMessages做成豆子：

// somewhere convenient...
@Bean
public List<String> errorMessages() {
return new ArrayList<>();
}

并将errorMessages注入到两个小任务中。

在CsvBatchReadertasklet 中，检查execute(...)引发的任何异常，并根据需要更新errorMessages：

public class CsvBatchReader<T> implements Tasklet, StepExecutionListener {
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
List<Throwable> failures = stepExecution.getFailureExceptions();
if (!failures.isEmpty())  {
errorMessages.add(...);
}
logger.info("Reader ended - " + clazz.getSimpleName());
return ExitStatus.COMPLETED;  // return INVALID if !failures.isEmpty() ?
}
}

既然errorMessages被注入CsvBatchReporting，那么：

public class CsvBatchReporting implements Tasklet, StepExecutionListener {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
logger.info("Reporting execute");
//Email Error 
String body = null;
String subject = "job finished: ";
if (!errorMessages.isEmpty())  {
subject += "ERROR";
body = ... // from errorMessages
} else {
subject += "SUCCESS";
}
return RepeatStatus.FINISHED;
}
}

Answer 3

虽然我最初的问题是关于将异常从一个步骤传递到下一个步骤，但我想指出一种使用exitStatus的替代方法

在我的阅读器步骤中，我可以捕获我的 csvToBean 异常并创建这样的存在状态。

@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws UnexpectedJobExecutionException {
logger.info("Reader execute - " + clazz.getSimpleName());
ICsvToBean csvToBean = new CsvToBean(clazz, path);
try {
batch = csvToBean.readCsv();
} catch(IOException ex) {
chunkContext.getStepContext().getStepExecution().setExitStatus(new ExitStatus("FAILED", ex.getMessage()));
}
return RepeatStatus.FINISHED;
}

然后，我会像这样在我的工作中创建条件流。

@Bean
public Job primaryCareJob(@Qualifier("reportingStep") Step reportingStep, @Qualifier("crossValidationStep") Step crossValidationStep) {
logger.info("Start PrimaryCare Job");
return jobs.get("primaryCareJob")
.start(readPrimaryCareStep()).on("FAILED").to(reportingStep)
.from(readPrimaryCareStep()).on("*").to(readPrimaryCareDetailStep())
.from(readPrimaryCareDetailStep()).on("FAILED").to(reportingStep)
.from(readPrimaryCareDetailStep()).on("*").to(processPrimaryCareStep())
.from(processPrimaryCareStep()).on("INVALID").to(reportingStep)
.from(processPrimaryCareStep()).on("*").to(processPrimaryCareDetailStep())
.from(processPrimaryCareDetailStep()).on("INVALID").to(reportingStep)
//Other steps
.from(reportingStep).on("*").end()
.from(reportingStep).on("*").fail()
.build()
.build();
}

最后，在我的读者任务步骤中，我将像这样检索现有状态

@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
JobExecution jobExecution = chunkContext.getStepContext().getStepExecution().getJobExecution();
Collection<StepExecution> stepExecutions = jobExecution.getStepExecutions();
Map<String, String> result = stepExecutions.stream()
.filter(x -> x.getExitStatus().getExitCode().equals("FAILED"))
.collect(Collectors.toMap(StepExecution::getStepName, x -> x.getExitStatus().getExitDescription()));
result.forEach((k, v) -> {
System.out.println("Step " + k + " failure " + v);
});
}