FlatFileFooterCallback异常无法进入SkipListener致Job终止求解决方案

问题描述 #

在Spring Batch任务中，FlatFileFooterCallback的writeFooter方法抛出异常时，无法被配置的SkipListener捕获，导致任务（Job）直接终止，无法继续执行。

核心原因 #

Spring Batch的skip机制和SkipListener仅覆盖Chunk内的Item读写处理环节（即reader.read()、processor.process()、writer.write(List)中的异常）。而FlatFileFooterCallback.writeFooter()是在Writer完成所有Chunk处理后、资源关闭前执行的收尾操作，不属于Chunk容错的覆盖范围，因此异常会直接向上传播导致Job终止。

解决方案 #

1. 修正设计：将数据库操作移至StepExecutionListener

由于FlatFileFooterCallback的核心职责是生成文件的footer内容，而非执行批量数据库插入，将该逻辑迁移到StepExecutionListener的afterStep方法是更合理的设计，同时能自主控制异常处理：

步骤1：实现StepExecutionListener

@Component
public class MMSSummaryAfterStepHandler implements StepExecutionListener {

    @Autowired
    @Qualifier("jdbcTemplate")
    private JdbcTemplate jdbcTemplate;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        // 无需实现，留空即可
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        ExecutionContext executionContext = stepExecution.getExecutionContext();
        HashMap<String, MMSDto> summaryMap = (HashMap<String, MMSDto>) executionContext.get("Summary");
        List<MMSDto> summaryValues = new ArrayList<>(summaryMap.values());

        try {
            jdbcTemplate.batchUpdate(InterConnectUtils.SUMMARY_EVENT_INSERT, new BatchPreparedStatementSetter() {
                @Override
                public void setValues(PreparedStatement ps, int i) throws SQLException {
                    MMSDto mmsDto = summaryValues.get(i);
                    ps.setString(1, mmsDto.getBatchNumber());
                    ps.setString(2, mmsDto.getEntityCode());
                    ps.setString(3, mmsDto.getInputFileType());
                    ps.setInt(4, mmsDto.getTotalType());
                    ps.setString(5, mmsDto.getServiceType());
                    ps.setString(6, mmsDto.getCallInd());
                    ps.setString(7, mmsDto.getCallDateTime());
                    ps.setString(8, mmsDto.getEventDirection());
                    ps.setString(9, mmsDto.getProductID());
                    ps.setInt(10, mmsDto.getTargetInd());
                    ps.setInt(11, mmsDto.getNoRecords());
                    ps.setLong(26, mmsDto.getTotalDuration());
                }

                @Override
                public int getBatchSize() {
                    return summaryMap.size();
                }
            });
            return stepExecution.getExitStatus();
        } catch (Exception e) {
            // 复用SkipListener的错误记录逻辑
            recordFooterError(summaryValues, e);
            // 标记Step状态为失败，但不终止Job
            return ExitStatus.FAILED.addExitDescription("汇总数据插入失败：" + e.getMessage());
        }
    }

    private void recordFooterError(List<MMSDto> summaryValues, Throwable t) {
        Exception ex = new Exception(t);
        for (MMSDto mmsDto : summaryValues) {
            jdbcTemplate.update(InterConnectUtils.BATCH_ERROR_INSERT, ps -> {
                StringWriter sw = new StringWriter();
                ex.printStackTrace(new PrintWriter(sw));
                String exceptionAsString = sw.toString();
                ps.setDate(1, new java.sql.Date(new Date().getTime()));
                ps.setString(2, mmsDto.getBatchNumber());
                ps.setString(3, "FOOTER_SUMMARY");
                Reader reader = new StringReader(exceptionAsString + "--" + ex.getMessage());
                ps.setClob(4, reader);
            });
        }
    }
}

步骤2：修改Step配置
移除FlatFileItemWriter中的footerCallback配置，并注册StepExecutionListener：

@Bean
@StepScope
public FlatFileItemWriter<BaseDTO> mmsItemWriter(@Value("#{stepExecutionContext['fileName']}") String filePath) throws Exception {
    BeanWrapperFieldExtractor<BaseDTO> fieldExtractor = new BeanWrapperFieldExtractor<>();
    fieldExtractor.setNames(mmsWriterFields);
    fieldExtractor.afterPropertiesSet();
    FormatterLineAggregator<BaseDTO> lineAggregator = new FormatterLineAggregator<>();
    lineAggregator.setFormat(mmsWriterFieldsFormat);
    lineAggregator.setFieldExtractor(fieldExtractor);
    String fileName = FilenameUtils.getName(filePath);
    Resource outputResource = new FileSystemResource(new File(sucessPath + fileName + inputDataSource));
    
    return new FlatFileItemWriterBuilder<BaseDTO>()
            .name("mmsWriter")
            .resource(outputResource)
            .lineAggregator(lineAggregator)
            .shouldDeleteIfEmpty(true)
            .build(); // 移除footerCallback配置
}

@Bean
public Step mmsSlaveStep() throws Exception {
    return stepBuilderFactory.get("mmsSlaveStep")
            .<BaseDTO, BaseDTO>chunk(chunkSize)
            .reader(mmsReader(null))
            .processor(mmsProcessor())
            .writer(mmsItemWriter(null))
            .faultTolerant()
            .skipLimit(skipErrorCount)
            .skip(Exception.class)
            .listener(itemSkipListener())
            .listener(new MMSSummaryAfterStepHandler()) // 注册新的Listener
            .build();
}

2. 临时规避：在writeFooter中手动捕获异常

如果不想调整现有结构，可以在MMSSummaryWriter的writeFooter方法中直接捕获异常，避免抛出导致Job终止：

@Override
public void writeFooter(Writer writer) throws IOException {
    ExecutionContext executionContext = stepExecution.getExecutionContext();
    HashMap<String, MMSDto> summaryMap = (HashMap<String, MMSDto>) executionContext.get("Summary");
    List<MMSDto> summaryValues = new ArrayList<>(summaryMap.values());

    try {
        jdbcTemplate.batchUpdate(InterConnectUtils.SUMMARY_EVENT_INSERT, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                MMSDto mmsDto = summaryValues.get(i);
                ps.setString(1, mmsDto.getBatchNumber());
                ps.setString(2, mmsDto.getEntityCode());
                ps.setString(3, mmsDto.getInputFileType());
                ps.setInt(4, mmsDto.getTotalType());
                ps.setString(5, mmsDto.getServiceType());
                ps.setString(6, mmsDto.getCallInd());
                ps.setString(7, mmsDto.getCallDateTime());
                ps.setString(8, mmsDto.getEventDirection());
                ps.setString(9, mmsDto.getProductID());
                ps.setInt(10, mmsDto.getTargetInd());
                ps.setInt(11, mmsDto.getNoRecords());
                ps.setLong(26, mmsDto.getTotalDuration());
            }

            @Override
            public int getBatchSize() {
                return summaryMap.size();
            }
        });
    } catch (Exception e) {
        // 记录错误到数据库
        recordFooterError(summaryValues, e);
        // 可选：标记Step执行状态
        stepExecution.setExitStatus(ExitStatus.FAILED.addExitDescription("Footer汇总插入失败"));
        // 不抛出异常，Job会继续执行
    }
}

// 复用错误记录逻辑
private void recordFooterError(List<MMSDto> summaryValues, Throwable t) {
    Exception ex = new Exception(t);
    for (MMSDto mmsDto : summaryValues) {
        jdbcTemplate.update(InterConnectUtils.BATCH_ERROR_INSERT, ps -> {
            StringWriter sw = new StringWriter();
            ex.printStackTrace(new PrintWriter(sw));
            String exceptionAsString = sw.toString();
            ps.setDate(1, new java.sql.Date(new Date().getTime()));
            ps.setString(2, mmsDto.getBatchNumber());
            ps.setString(3, "FOOTER_WRITER");
            Reader reader = new StringReader(exceptionAsString + "--" + ex.getMessage());
            ps.setClob(4, reader);
        });
    }
}

内容的提问来源于stack exchange，提问作者Rakesh