Dynamic chunk size via CompletionPolicy no longer possible in new Chunk API?

[marbon87]

Hi,

in the old chunk implementation it was possible to configure a dynamic chunk size per job execution by using a CompletionPolicy that could read a chunk-size value from job parameters.

For example:

        @Bean
        public Step step(JobRepository jobRepository,
                         SimpleCompletionPolicy completionPolicy) {
            return new StepBuilder("step", jobRepository)
                    .<Integer, Integer>chunk(completionPolicy, new ResourcelessTransactionManager())
                    ...
                    .build();
        }

        @Bean
        @StepScope
        public SimpleCompletionPolicy chunkCompletionPolicy() {
            return new SimpleCompletionPolicy(2); // Read from job parameters
        }

With the new chunk-oriented programming model, the chunk() method only accepts a fixed int value and does not offer an overload that accepts a CompletionPolicy.
However, the CompletionPolicy class itself is not deprecated, which raises the question:
Is dynamic chunk sizing officially no longer supported in the new API, or is a chunk method that accepts a CompletionPolicy simply missing from the new StepBuilder? Or is there another way to solve this?

Regards,
Mark

[fmbenhassine]

CompletionPolicy is part of the repeat framework, which is not used in the new implementation. That package is not deprecated (only its concurrency feature is deprecated) in case a custom step implementation wants to use it.

Defining the chunk size dynamically could be done by defining a step-scoped tasklet, not necessarily through a step-scoped completion policy (this is true for the old implementation in v5 as well). There is a note about this here, but I can provide an example if needed.

[marbon87]

Hi @fmbenhassine,

thanks for your response.
Regarding to your feedback in #3900 (comment) i thought, that using @JobScope or @StepScope on a Step-Bean is no adviced although it's working in this situtation.
Is this statement still relevant for the new chunk-processing implementation?

Maybe an example would help to resolve the confusion

[fmbenhassine]

I get your point: In v6, there is no tasklet to scope, so you are wondering how to do that now.

Defining the chunk size as a step-scoped integer bean won't work (as Integer is a final class and can't be subclassed to create a step-scoped proxy).

Regarding to your feedback in #3900 (comment) i thought, that using https://github.com/JobScope or @StepScope on a Step-Bean is no adviced although it's working in this situtation.
Is this statement still relevant for the new chunk-processing implementation?

I guess no, and I need to reconsider my thoughts around the job scope. Since completion policy is not used anymore, and defining a step-scoped integer bean won't work, I think the only way to do that in v6 is by using a job-scoped step bean (in which case, we might have found a valid use case for job-scope starting from v6, which invalidates this discussion: #4869, and as a consequence, we would need to update the documentation accordingly for v6).

Thoughts?

[fmbenhassine]

As for the example, here is a quick one with a job-scoped step:

package org.springframework.batch.samples.helloworld;

import java.util.List;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.job.parameters.JobParameters;
import org.springframework.batch.core.job.parameters.JobParametersBuilder;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.Step;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.infrastructure.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class HelloWorldJobConfiguration {

	@Bean
	@JobScope
	public Step step(JobRepository jobRepository, @Value("#{jobParameters['chunkSize']}") int chunkSize) {
		return new StepBuilder(jobRepository).
				<Integer, Integer>chunk(chunkSize)
				.reader(new ListItemReader<>(List.of(1, 2, 3, 4, 5)))
				.writer(items -> {
					System.out.println("Writing chunk: " + items);
                    items.forEach(System.out::println);
                })
				.build();
	}

	@Bean
	public Job job(JobRepository jobRepository, Step step) {
		return new JobBuilder(jobRepository).start(step).build();
	}

	public static void main(String[] args) throws Exception {
		ApplicationContext context = new AnnotationConfigApplicationContext(HelloWorldJobConfiguration.class);
		JobOperator jobOperator = context.getBean(JobOperator.class);
		Job job = context.getBean(Job.class);

		// when
		JobParameters jobParameters = new JobParametersBuilder()
				.addLong("chunkSize", 2L) // should result in 3 chunks: [1, 2], [3, 4], [5]
				.toJobParameters();
		jobOperator.start(job, jobParameters);
	}

}

prints

16:56:15.471 [main] INFO  o.s.b.c.l.s.TaskExecutorJobLauncher - Job: [SimpleJob: [name=job]] launched with the following parameters: [{JobParameter{name='chunkSize', value=2, type=class java.lang.Long, identifying=true}}]
16:56:15.507 [main] INFO  o.s.batch.core.job.SimpleStepHandler - Executing step: [scopedTarget.step]
Writing chunk: [items=[1, 2], skips=[]]
1
2
Writing chunk: [items=[3, 4], skips=[]]
3
4
Writing chunk: [items=[5], skips=[]]
5
16:56:15.513 [main] INFO  o.s.batch.core.step.AbstractStep - Step: [scopedTarget.step] executed in 5ms
16:56:15.514 [main] INFO  o.s.b.c.l.s.TaskExecutorJobLauncher - Job: [SimpleJob: [name=job]] completed with the following parameters: [{JobParameter{name='chunkSize', value=2, type=class java.lang.Long, identifying=true}}] and the following status: [COMPLETED] in 38ms

[marbon87]

Sounds good to me. But i do not remember why job-scoped step bean were not adviced in the documentation before. I guess there was a technical reason for that up to now, right?

[fmbenhassine]

The reason is that prior to v6, where the implementation of the chunk-oriented processing model was based on Tasklet / TaskletStep, it was possible to scope the tasklet itself to the step, meaning there was an "intermediate" batch artefact that can be step-scoped (therefore no need for a job scope). Now with v6, the chunk-oriented processing model is directly implemented in a Step implementation, thus, the intermediate concept is gone. So indeed, the reason is purely technical in this case.

I will close #4869 with a link to this discussion and I will update the reference documentation accordingly. Once that done, I will close this discussion. Of course if you need more details on this, please let me know and I will be happy to help.

[fmbenhassine]

I will update the reference documentation accordingly

Done: 94d744f