Spring批處理集成Java DSL和RunIdIncrementer不增加

Question

我有一個spring啓動/集成/批處理，它將運行並輪詢SFTP上的文件。

我希望能夠使用在Job中定義的RunIdIncrementer來重新啓動一個作業（可能是因爲應用程序已重新啓動，或者因爲某些原因我們再次收到相同的文件）具有相同參數（基本上是相同文件）組態。

不幸的是run.id = 1不遞增，我得到一個JobInstanceAlreadyCompleteException

作業配置

@Autowired 
private JobBuilderFactory jobBuilders; 

@Bean 
public Job importOffersJob() { 

    Job job = jobBuilders.get("importOffersJob") 
      .start(importOffersStep) 
      .listener(traceJobExecutionListener()) 
      .incrementer(new RunIdIncrementer()) 
      .build(); 

    return job; 
} 

集成

@Bean 
public IntegrationFlow ftpInboundFlow() { 
    return IntegrationFlows 
      .from(Sftp.inboundAdapter(SftpSessionFactory()) 
       .regexFilter(".*\\.xml.mini$") 
       .deleteRemoteFiles(intCfg.getSftpDeleteRemoteFiles()) 
       .preserveTimestamp(Boolean.TRUE) 
       .autoCreateLocalDirectory(Boolean.TRUE) 
       .remoteDirectory(intCfg.getSftpRemoteDirectory()) 
       .localDirectory(new File(intCfg.getSftpLocalDirectory()) 
      ), 
       e -> e.id("sftpInboundAdapter") 
       .poller(Pollers.fixedRate(intCfg.getSftpPollerInMinutes(), TimeUnit.MINUTES).maxMessagesPerPoll(1))) 
      .transform(fileToJobLaunchRequestTransformer())   
      .handle(jobLaunchingGw()) 
      .handle(logger()) 
      .get(); 
} 

---

public class FileToJobLaunchRequestTransformer implements GenericTransformer<Message<File>, JobLaunchRequest> { 

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

    @Autowired 
    @Qualifier("importOffersJob") 
    private Job job; 

    @Override 
    public JobLaunchRequest transform(Message<File> source) { 
     log.info("FileToJobLaunchRequestTransformer, source.getPayload().getAbsolutePath(): {}", source.getPayload() 
       .getAbsolutePath()); 

     JobParametersBuilder jobParametersBuilder = new JobParametersBuilder(); 
     jobParametersBuilder.addString("pathToFile", "file:" + source.getPayload().getAbsolutePath()); 
     //jobParametersBuilder.addString("UUID", UUID.randomUUID().toString()); 

     JobParameters jobParams = job.getJobParametersIncrementer().getNext(jobParametersBuilder.toJobParameters()); 

     return new JobLaunchRequest(job, jobParams); 
     } 
    } 

如果我取消註釋jobParametersBuilder.addString("UUID", UUID.randomUUID().toString());這是工作，但我認爲重點是能夠重用在我的作業配置中定義的增量器是不是？

（當我正在運行的同批爲沒有整合增量正在一個簡單的彈簧引導）

千恩萬謝

Answer 1

RunIdIncrementer.getNext()：

public JobParameters getNext(JobParameters parameters) { 

    JobParameters params = (parameters == null) ? new JobParameters() : parameters; 

    long id = params.getLong(key, 0L) + 1; 
    return new JobParametersBuilder(params).addLong(key, id).toJobParameters(); 
} 

您正在創建一個新的作業參數每次這樣它總是會返回1，因爲run.id不存在。

如果將jobParametersBuilder到現場變壓器，旁邊job，它會工作，但僅限於這個實例您的應用程序中。它會在下一次啓動您的應用時從1開始。

要重新生存，您需要在某處保存run.id值，或者需要從存儲庫中獲取最後的作業參數。