title: How can you restart a failed spring batch job and let it pick up where it left off?
body: <p>According to the Spring Batch documentation restarting of a job is supported out of the box but I cannot get it to start from where it left of. e.g. If my step processed 10 records it should start at record 11 with processing whenever I restart it. In practice this doesn't happen. It reads from the beginnen en reprocesses everything. </p>
  
  <p>Does anybody have a Java config based configuration of a simple job that reads a delimited file and writes the content to a db table that can be restarted from the point it stopped?</p>
  
  <pre><code>@Configuration
  public class BatchConfiguration {
  
      @Value("${spring-batch.databaseType}")
      private String databaseType;
  
      @Value("${spring-batch.databaseSchema}")
      private String schemaName;
  
      @Bean
      public JobBuilderFactory jobBuilderFactory(final JobRepository jobRepository) {
          return new JobBuilderFactory(jobRepository);
      }
  
      @Bean
      public StepBuilderFactory stepBuilderFactory(final JobRepository jobRepository,
          final PlatformTransactionManager transactionManager) {
          return new StepBuilderFactory(jobRepository, transactionManager);
      }
  
      @Bean
      public JobRepository jobRepository(final DataSource dataSource, final PlatformTransactionManager transactionManager) {
  
          final JobRepositoryFactoryBean bean = new JobRepositoryFactoryBean();
          bean.setDatabaseType(databaseType);
          bean.setDataSource(dataSource);
          if (StringUtils.isNotBlank(schemaName)) {
              bean.setTablePrefix(schemaName);
          }
          bean.setTransactionManager(transactionManager);
          try {
              bean.afterPropertiesSet();
              return bean.getObject();
          } catch (final Exception e) {
              throw new BatchConfigurationException("Invalid batch job repository configuration.", e);
          }
      }
  
      @Bean
      public JobLauncher jobLauncher(final JobRepository jobRepository) {
  
          final SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
          jobLauncher.setJobRepository(jobRepository);
          return jobLauncher;
      }
  
  }
  
  @Configuration
  @EnableScheduling
  @ComponentScan("com.some.package")
  public class BatchJobConfiguration {
  
      @Resource
      private JobBuilderFactory jobBuilderFactory;
  
      @Resource
      private StepBuilderFactory stepBuilderFactory;
  
      @Value("${savings-transaction.file}")
      private String savingsTransactionFile;
  
      @Value("${savings-balance.file}")
      private String savingsBalanceFile;
  
      @Value("${processed-directory}")
      private String processedDirectory;
  
      private static final Integer IMPORT_CHUNKSIZE = 10;
  
      @Bean
      @DependsOn("stepBuilderFactory")
      public Step savingsTransactionStep(final PlatformTransactionManager transactionManager,
              @Qualifier("savingsTransactionItemReader") final ItemReader&lt;SavingsTransactionItem&gt; savingsTransactionItemReader,
              @Qualifier("savingsTransactionProcessor") final ItemProcessor&lt;SavingsTransactionItem, SavingsTransaction&gt; processor,
              @Qualifier("savingsTransactionItemWriter") final ItemWriter&lt;SavingsTransaction&gt; savingsTransactionItemWriter,
              @Qualifier("savingsTransactionStepListener") final SavingsTransactionStepListener listener) {
  
          return stepBuilderFactory.get("savingsTransactionStep")
                  .transactionManager(transactionManager)
                  .&lt;SavingsTransactionItem, SavingsTransaction&gt; chunk(IMPORT_CHUNKSIZE)
                  .reader(savingsTransactionItemReader)
                  .processor(processor)
                  .writer(savingsTransactionItemWriter)
                  .listener(listener)
                  .build();
      }
  
      @Bean
      public Step savingsTransactionCleanUpStep(final PlatformTransactionManager transactionManager,
              final JobRepository jobRepository) {
  
          final TaskletStep taskletStep = new TaskletStep("savingsTransactionCleanUpStep");
  
          final FileMovingTasklet tasklet = new FileMovingTasklet();
          tasklet.setFileNamePattern(savingsTransactionFile);
          tasklet.setProcessedDirectory(processedDirectory);
          taskletStep.setTasklet(tasklet);
          taskletStep.setTransactionManager(transactionManager);
          taskletStep.setJobRepository(jobRepository);
          try {
              taskletStep.afterPropertiesSet();
          } catch (final Exception e) {
              throw new BatchConfigurationException("Failed to configure tasklet!", e);
          }
  
          return taskletStep;
      }
  
      @Bean
      @DependsOn("jobBuilderFactory")
      public Job job(final Step savingsTransactionStep,
              final Step savingsTransactionCleanUpStep) {
  
          return jobBuilderFactory.get("job")
                  .incrementer(new RunIdIncrementer())
                  .start(savingsTransactionStep)  
                  .next(savingsTransactionCleanUpStep)                    
                  .on("FINISHED")
                  .end()
                  .build()
                  .build();
      }
  }
  </code></pre>
  
  <p>Unit test code that restarts the job</p>
  
  <pre><code>    final Date now = new Date();
      jobMananger.processRegistrations(now);
  
      final List&lt;SavingsBalance&gt; savingsBalances = savingsBalanceDao.findAll();
      assertEquals(9, savingsBalances.size());
  
      FileUtils.moveFile(new File("target/AEA001_20160610.dat"), new File("target/AEA001_20160610_invalid.dat"));
      FileUtils.moveFile(new File("target/AEA001_20160610_valid.dat"), new File("target/AEA001_20160610.dat"));
  
      jobMananger.processRegistrations(now);
  
      final List&lt;SavingsBalance&gt; savingsBalances2 = savingsBalanceDao.findAll();
      System.out.println(savingsBalances2.size());
      int found = 0;
      for (final SavingsBalance savingsBalance : savingsBalances2) {
  
          final String id = savingsBalance.getId();
          if ("12345".equals(id)) {
              found++;
          }
  
      }
  
      assertEquals("Invalid number of found balances!", 1, found);
  </code></pre>
  
  <p>The job manager implementation</p>
  
  <pre><code>public class JobManager {
  
      @Resource
      private JobLauncher jobLauncher;
  
      @Resource
      private Job job;
  
      @Transactional(propagation = Propagation.NOT_SUPPORTED)
      public void processRegistrations(final Date date) {
  
          try {
  
              final Map&lt;String, JobParameter&gt; parameters = new HashMap&lt;&gt;();
              parameters.put("START_DATE", new JobParameter(date));
  
              final JobParameters jobParameters = new JobParameters(parameters);
              final JobExecution execution = jobLauncher.run(job, jobParameters);
              LOG.info("Exit Status : " + execution.getStatus());
          } catch (JobExecutionAlreadyRunningException | JobRestartException | JobInstanceAlreadyCompleteException
                  | JobParametersInvalidException e) {
              LOG.error("Failed to process registrations.", e);
          }
      }
  
  }
  </code></pre>