容错spring批处理作业读取少量记录后卡顿

Question

Im使用存储库项目读取器从数据库读取事务，使用flatfileitem写入器处理和写入文件。我使用容错跳过任何记录，如果它是错误的。为了检查一条记录是否有错误，我们在处理器中进行一些验证并抛出自定义异常。

批处理配置

return stepBuilderFactory.get("test")
        .<Transaction, Transaction>chunk(1)
        .reader(this.reader.read())
        .processor(transactionProcessor)
        .writer(transactionWriter)
        .faultTolerant()
        .skipPolicy(new UnlimitedSkipPolicy()
        .listener(new TransactionSkipListenerTransaction())
        .stream(this.transactionwriter)

事务读取器bean

 @Override
@StepScope
public RepositoryItemReader<Transaction> read() throws Exception {
    final ZonedDateTime zonedDateTime = LocalDate.now().atStartOfDay(ZoneId.systemDefault());
    final Date today = Date.from(zonedDateTime.toInstant());
    final Date yesterday = Date.from(zonedDateTime.minusDays(1L).toInstant());
    RepositoryItemReader<Transaction> reader = new RepositoryItemReader<>();
    reader.setRepository(transactionDao);
    reader.setMethodName("findByTransactionDateGreaterThanEqualAndTransactionDateLessThan");
    reader.setArguments(Arrays.asList(yesterday, today));
    reader.setSort(Collections.singletonMap("transactionId", Sort.Direction.ASC));
    reader.setPageSize(10000);
    return reader;
}


Processor bean 
@Override
public Transaction process(Transaction transaction) throws Exception {
if(utility.isInvalidTransaction(transaction)) {
  log.error("Invalid transaction {} and skipped from reporting", transaction.getTransactionId());
  throw new CustomException("invalid transaction" + transaction.getTransactionid());
}

而writer只是一个flatfile编写器。问题是，在读取了很少事务之后，在我的例子中，在读取并处理了9个无效事务之后，批处理作业被卡住了。不知道为什么。

.reader(this.reader.read())
        .processor(transactionProcessor)
        .writer(transactionWriter)
        .faultTolerant()
        .skipPolicy(new UnlimitedSkipPolicy()
        .noRetry(Exception.class)
        .noRollBack(Exception.class)

困在这里

Answer 1

在我看来，processbean的职责是添加自定义转换或逻辑，在此基础上将数据传递给编写器。在这种情况下，我只是记录错误并返回null，在编写器处理null对象引用的情况下，即无效事务(null)没有传递给编写器。