jonatan-ivanov/BatchConfiguration.java

## Readme.md

      
    Raw
  

              Readme.md
            
          
I have a paginated REST API which I would like to poll
Since the source data is growing, I would like to set this up as a scheduled sync job
and rerun it from the point where the previous execution left off (even if it was COMPLETED)
basically advancing on the new data periodically
AbstractPaginatedDataItemReader seems to be suitable, it is able to:

Track the read item count (actually it is AbstractItemCountingItemStreamItemReader)
Maintain and persist its state in the ExecutionContext (read.count)


What I'm missing is creating the new JobParameters based on the previous ExecutionContext
It seems DefaultJobParametersExtractor does the trick I need
And it is used from the JobStep class in a way I need
Unfortunately this is only available to JobStep and I have a TaskletStep
So I ended up reimplementing the parts I need from JobStep and DefaultJobParametersExtractor

Is there a better way to do this?
The code in Example.java is what I would like to get rid off by providing the RunIdIncrementer and a JobParametersExtractor.

  
## BatchConfiguration.java
@Configuration
@EnableScheduling
@EnableBatchProcessing
public class BatchConfiguration {
  @Bean
  @StepScope
  public ResourceReader reader(
      @Value("...") String url,
      @Value("...") int pageSize,
      @Value("#{jobParameters['ResourceReader.index']}") int index) {
    return new ResourceReader(url, pageSize, index);
  }

  @Bean
  public Job feedingJob(JobBuilderFactory jobBuilderFactory, Step feedingStep) {
    return jobBuilderFactory.get("feedingJob")
        .incrementer(new RunIdIncrementer())
        .flow(feedingStep)
        .end()
        .build();
  }

  @Bean
  public Step feedingStep(StepBuilderFactory stepBuilderFactory, ResourceReader reader) {
    return stepBuilderFactory.get("feedingStep")
        .<Resource, String>chunk(10)
        .reader(reader)
        .processor((Function<Resource, String>) (Resource::getId))
        .writer(System.out::println)
        .build();
  }

  @Bean
  public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry) {...}
}

## Example.java
public class Example {
  private final JobLauncher jobLauncher;
  private final JobExplorer jobExplorer;
  private final Job job;

  @Scheduled(fixedDelayString = "${...}")
  public void run() throws Exception {
    jobLauncher.run(job, createJobParameters());
  }

  private JobParameters createJobParameters() {
    return Optional.of(jobExplorer.getJobInstances(job.getName(), 0, 1))
        .filter(jobInstances -> !jobInstances.isEmpty())
        .map(jobInstances -> jobExplorer.getJobExecutions(jobInstances.get(0)))
        .filter(jobExecutions -> !jobExecutions.isEmpty())
        .map(jobExecutions -> jobExecutions.get(0).getStepExecutions())
        .filter(stepExecutions -> !stepExecutions.isEmpty())
        .map(stepExecutions -> Iterables.get(stepExecutions, 0))
        .map(stepExecution -> createJobParameters(stepExecution.getExecutionContext()))
        .orElseGet(() -> createJobParameters(0));
  }

  private JobParameters createJobParameters(ExecutionContext executionContext) {
    return createJobParameters(job, executionContext.getLong("ResourceReader.index"));
  }

  private JobParameters createJobParameters(long index) {
    return new JobParametersBuilder(jobExplorer)
        .getNextJobParameters(job)
        .addLong("ResourceReader.index", index) // no addInt, though the index must be an int because of the abstract reader
        .toJobParameters();
  }
}

## ResourceReader.java
public class ResourceReader extends AbstractPaginatedDataItemReader<Resource> {
  private final String url;

  public ResourceReader(String url, int pageSize, int index) {
    this.url = url;
    this.pageSize = pageSize;
    setCurrentItemCount(index);
    setName(this.getClass().getSimpleName());
  }

  @Override
  protected Iterator<Resource> doPageRead() {
    return getResources(getIndex(), pageSize).iterator();
  }

  private List<Resource> getResources(int index, int limit) {
    // HTTP call to a paginated API
  }

  @Override
  protected void jumpToItem(int itemLastIndex) {
    // the API is paginated, no need to jump
  }

  @Override
  public void update(ExecutionContext executionContext) throws ItemStreamException {
    super.update(executionContext);

    // the current item count and the index are int but the JobParameters only have addLong there is no addInt
    executionContext.putLong(getExecutionContextKey("index"), getIndex());
  }

  private int getIndex() {
    return getCurrentItemCount() - 1;
  }
}
	@Configuration
	@EnableScheduling
	@EnableBatchProcessing
	public class BatchConfiguration {
	@Bean
	@StepScope
	public ResourceReader reader(
	@Value("...") String url,
	@Value("...") int pageSize,
	@Value("#{jobParameters['ResourceReader.index']}") int index) {
	return new ResourceReader(url, pageSize, index);
	}

	@Bean
	public Job feedingJob(JobBuilderFactory jobBuilderFactory, Step feedingStep) {
	return jobBuilderFactory.get("feedingJob")
	.incrementer(new RunIdIncrementer())
	.flow(feedingStep)
	.end()
	.build();
	}

	@Bean
	public Step feedingStep(StepBuilderFactory stepBuilderFactory, ResourceReader reader) {
	return stepBuilderFactory.get("feedingStep")
	.<Resource, String>chunk(10)
	.reader(reader)
	.processor((Function<Resource, String>) (Resource::getId))
	.writer(System.out::println)
	.build();
	}

	@Bean
	public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry) {...}
	}
	public class Example {
	private final JobLauncher jobLauncher;
	private final JobExplorer jobExplorer;
	private final Job job;

	@Scheduled(fixedDelayString = "${...}")
	public void run() throws Exception {
	jobLauncher.run(job, createJobParameters());
	}

	private JobParameters createJobParameters() {
	return Optional.of(jobExplorer.getJobInstances(job.getName(), 0, 1))
	.filter(jobInstances -> !jobInstances.isEmpty())
	.map(jobInstances -> jobExplorer.getJobExecutions(jobInstances.get(0)))
	.filter(jobExecutions -> !jobExecutions.isEmpty())
	.map(jobExecutions -> jobExecutions.get(0).getStepExecutions())
	.filter(stepExecutions -> !stepExecutions.isEmpty())
	.map(stepExecutions -> Iterables.get(stepExecutions, 0))
	.map(stepExecution -> createJobParameters(stepExecution.getExecutionContext()))
	.orElseGet(() -> createJobParameters(0));
	}

	private JobParameters createJobParameters(ExecutionContext executionContext) {
	return createJobParameters(job, executionContext.getLong("ResourceReader.index"));
	}

	private JobParameters createJobParameters(long index) {
	return new JobParametersBuilder(jobExplorer)
	.getNextJobParameters(job)
	.addLong("ResourceReader.index", index) // no addInt, though the index must be an int because of the abstract reader
	.toJobParameters();
	}
	}
	public class ResourceReader extends AbstractPaginatedDataItemReader<Resource> {
	private final String url;

	public ResourceReader(String url, int pageSize, int index) {
	this.url = url;
	this.pageSize = pageSize;
	setCurrentItemCount(index);
	setName(this.getClass().getSimpleName());
	}

	@Override
	protected Iterator<Resource> doPageRead() {
	return getResources(getIndex(), pageSize).iterator();
	}

	private List<Resource> getResources(int index, int limit) {
	// HTTP call to a paginated API
	}

	@Override
	protected void jumpToItem(int itemLastIndex) {
	// the API is paginated, no need to jump
	}

	@Override
	public void update(ExecutionContext executionContext) throws ItemStreamException {
	super.update(executionContext);

	// the current item count and the index are int but the JobParameters only have addLong there is no addInt
	executionContext.putLong(getExecutionContextKey("index"), getIndex());
	}

	private int getIndex() {
	return getCurrentItemCount() - 1;
	}
	}