Improve Step/Job builder APIs to guide users to set mandatory properties

Before this commit, the user had to manually set the job repository and transaction
manager on `JobBuilder` and `StepBuilder` instances with a chained call to
`.repository(jobRepository)` and `.transactionManager(transactionManager)`.
This is error prone and can lead to runtime errors if these properties are not set.

This commit introduces new APIs to guide the user to set these properties at
builder creation time.

Resolves #4192

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/JobBuilderFactory.java

@@ -51,7 +51,7 @@ public JobBuilderFactory(JobRepository jobRepository) {
 	 * @return a job builder
 	 */
 	public JobBuilder get(String name) {
-		return new JobBuilder(name).repository(this.jobRepository);
+		return new JobBuilder(name, this.jobRepository);
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/StepBuilderFactory.java

@@ -53,7 +53,7 @@ public StepBuilderFactory(JobRepository jobRepository) {
 	 * @return a step builder
 	 */
 	public StepBuilder get(String name) {
-		return new StepBuilder(name).repository(this.jobRepository);
+		return new StepBuilder(name, this.jobRepository);
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java

@@ -274,7 +274,6 @@ protected void enhanceCommonStep(StepBuilderHelper<?> builder) {
 		if (startLimit != null) {
 			builder.startLimit(startLimit);
 		}
-		builder.repository(jobRepository);
 		for (Object listener : stepExecutionListeners) {
 			if (listener instanceof StepExecutionListener) {
 				builder.listener((StepExecutionListener) listener);
@@ -290,10 +289,11 @@ protected Step createPartitionStep() {
 
 		PartitionStepBuilder builder;
 		if (partitioner != null) {
-			builder = new StepBuilder(name).partitioner(step != null ? step.getName() : name, partitioner).step(step);
+			builder = new StepBuilder(name, jobRepository)
+					.partitioner(step != null ? step.getName() : name, partitioner).step(step);
 		}
 		else {
-			builder = new StepBuilder(name).partitioner(step);
+			builder = new StepBuilder(name, jobRepository).partitioner(step);
 		}
 		enhanceCommonStep(builder);
 
@@ -401,7 +401,7 @@ else if (skipLimit != null) {
 	 * @return the {@link FaultTolerantStepBuilder}.
 	 */
 	protected FaultTolerantStepBuilder<I, O> getFaultTolerantStepBuilder(String stepName) {
-		return new FaultTolerantStepBuilder<>(new StepBuilder(stepName));
+		return new FaultTolerantStepBuilder<>(new StepBuilder(stepName, jobRepository));
 	}
 
 	protected void registerItemListeners(SimpleStepBuilder<I, O> builder) {
@@ -445,16 +445,16 @@ protected CompletionPolicy getCompletionPolicy() {
 	}
 
 	protected SimpleStepBuilder<I, O> getSimpleStepBuilder(String stepName) {
-		return new SimpleStepBuilder<>(new StepBuilder(stepName));
+		return new SimpleStepBuilder<>(new StepBuilder(stepName, jobRepository));
 	}
 
 	/**
 	 * Create a new {@link TaskletStep}.
 	 * @return a new {@link TaskletStep}
 	 */
 	protected TaskletStep createTaskletStep() {
-		TaskletStepBuilder builder = new TaskletStepBuilder(new StepBuilder(name))
-				.transactionManager(transactionManager).tasklet(tasklet);
+		TaskletStepBuilder builder = new TaskletStepBuilder(new StepBuilder(name, jobRepository)).tasklet(tasklet,
+				transactionManager);
 		enhanceTaskletStepBuilder(builder);
 		return builder.build();
 	}
@@ -512,14 +512,14 @@ public boolean rollbackOn(Throwable ex) {
 	 * @return the {@link org.springframework.batch.core.job.flow.FlowStep}.
 	 */
 	protected Step createFlowStep() {
-		FlowStepBuilder builder = new StepBuilder(name).flow(flow);
+		FlowStepBuilder builder = new StepBuilder(name, jobRepository).flow(flow);
 		enhanceCommonStep(builder);
 		return builder.build();
 	}
 
 	private Step createJobStep() throws Exception {
 
-		JobStepBuilder builder = new StepBuilder(name).job(job);
+		JobStepBuilder builder = new StepBuilder(name, jobRepository).job(job);
 		enhanceCommonStep(builder);
 		builder.parametersExtractor(jobParametersExtractor);
 		builder.launcher(jobLauncher);

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2013 the original author or authors.
+ * Copyright 2006-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,11 +17,13 @@
 
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.repository.JobRepository;
 
 /**
  * Convenience for building jobs of various kinds.
  *
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  * @since 2.2
  *
  */
@@ -30,11 +32,24 @@ public class JobBuilder extends JobBuilderHelper<JobBuilder> {
 	/**
 	 * Create a new builder for a job with the given name.
 	 * @param name the name of the job
+	 * @deprecated use {@link JobBuilder#JobBuilder(String, JobRepository)}
 	 */
+	@Deprecated(since = "5.0")
 	public JobBuilder(String name) {
 		super(name);
 	}
 
+	/**
+	 * Create a new builder for a job with the given name.
+	 * @param name the name of the job
+	 * @param jobRepository the job repository to which the job should report to
+	 * @since 5.0
+	 */
+	public JobBuilder(String name, JobRepository jobRepository) {
+		super(name);
+		super.repository(jobRepository);
+	}
+
 	/**
 	 * Create a new job builder that will execute a step or sequence of steps.
 	 * @param step a step to execute

spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java

@@ -19,35 +19,64 @@
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.job.flow.Flow;
 import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.repeat.CompletionPolicy;
+import org.springframework.transaction.PlatformTransactionManager;
 
 /**
  * Convenient entry point for building all kinds of steps. Use this as a factory for
  * fluent builders of any step.
  *
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  * @since 2.2
  */
 public class StepBuilder extends StepBuilderHelper<StepBuilder> {
 
 	/**
 	 * Initialize a step builder for a step with the given name.
 	 * @param name the name of the step
+	 * @deprecated use {@link StepBuilder#StepBuilder(String, JobRepository)}
 	 */
+	@Deprecated(since = "5.0")
 	public StepBuilder(String name) {
 		super(name);
 	}
 
+	/**
+	 * Initialize a step builder for a step with the given name and job repository.
+	 * @param name the name of the step
+	 * @param jobRepository the job repository to which the step should report to.
+	 * @since 5.0
+	 */
+	public StepBuilder(String name, JobRepository jobRepository) {
+		super(name);
+		super.repository(jobRepository);
+	}
+
 	/**
 	 * Build a step with a custom tasklet, not necessarily item processing.
 	 * @param tasklet a tasklet
 	 * @return a {@link TaskletStepBuilder}
+	 * @deprecated use {@link StepBuilder#tasklet(Tasklet, PlatformTransactionManager)}
 	 */
+	@Deprecated(since = "5.0")
 	public TaskletStepBuilder tasklet(Tasklet tasklet) {
 		return new TaskletStepBuilder(this).tasklet(tasklet);
 	}
 
+	/**
+	 * Build a step with a custom tasklet, not necessarily item processing.
+	 * @param tasklet a tasklet
+	 * @param transactionManager the transaction manager to use for the tasklet
+	 * @return a {@link TaskletStepBuilder}
+	 * @since 5.0
+	 */
+	public TaskletStepBuilder tasklet(Tasklet tasklet, PlatformTransactionManager transactionManager) {
+		return new TaskletStepBuilder(this).tasklet(tasklet, transactionManager);
+	}
+
 	/**
 	 * Build a step that processes items in chunks with the size provided. To extend the
 	 * step to being fault tolerant, call the {@link SimpleStepBuilder#faultTolerant()}
@@ -62,11 +91,35 @@ public TaskletStepBuilder tasklet(Tasklet tasklet) {
 	 * @return a {@link SimpleStepBuilder}
 	 * @param <I> the type of item to be processed as input
 	 * @param <O> the type of item to be output
+	 * @deprecated use {@link StepBuilder#chunk(int, PlatformTransactionManager)}
 	 */
+	@Deprecated(since = "5.0")
 	public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
 		return new SimpleStepBuilder<I, O>(this).chunk(chunkSize);
 	}
 
+	/**
+	 * Build a step that processes items in chunks with the size provided. To extend the
+	 * step to being fault tolerant, call the {@link SimpleStepBuilder#faultTolerant()}
+	 * method on the builder. In most cases you will want to parameterize your call to
+	 * this method, to preserve the type safety of your readers and writers, e.g.
+	 *
+	 * <pre>
+	 * new StepBuilder(&quot;step1&quot;).&lt;Order, Ledger&gt; chunk(100, transactionManager).reader(new OrderReader()).writer(new LedgerWriter())
+	 * // ... etc.
+	 * </pre>
+	 * @param chunkSize the chunk size (commit interval)
+	 * @param transactionManager the transaction manager to use for the chunk-oriented
+	 * tasklet
+	 * @return a {@link SimpleStepBuilder}
+	 * @param <I> the type of item to be processed as input
+	 * @param <O> the type of item to be output
+	 * @since 5.0
+	 */
+	public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize, PlatformTransactionManager transactionManager) {
+		return new SimpleStepBuilder<I, O>(this).transactionManager(transactionManager).chunk(chunkSize);
+	}
+
 	/**
 	 * Build a step that processes items in chunks with the completion policy provided. To
 	 * extend the step to being fault tolerant, call the
@@ -81,12 +134,39 @@ public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
 	 * @param completionPolicy the completion policy to use to control chunk processing
 	 * @return a {@link SimpleStepBuilder}
 	 * @param <I> the type of item to be processed as input
-	 * @param <O> the type of item to be output *
+	 * @param <O> the type of item to be output
+	 * @deprecated use
+	 * {@link StepBuilder#chunk(CompletionPolicy, PlatformTransactionManager)}
 	 */
+	@Deprecated(since = "5.0")
 	public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
 		return new SimpleStepBuilder<I, O>(this).chunk(completionPolicy);
 	}
 
+	/**
+	 * Build a step that processes items in chunks with the completion policy provided. To
+	 * extend the step to being fault tolerant, call the
+	 * {@link SimpleStepBuilder#faultTolerant()} method on the builder. In most cases you
+	 * will want to parameterize your call to this method, to preserve the type safety of
+	 * your readers and writers, e.g.
+	 *
+	 * <pre>
+	 * new StepBuilder(&quot;step1&quot;).&lt;Order, Ledger&gt; chunk(100, transactionManager).reader(new OrderReader()).writer(new LedgerWriter())
+	 * // ... etc.
+	 * </pre>
+	 * @param completionPolicy the completion policy to use to control chunk processing
+	 * @param transactionManager the transaction manager to use for the chunk-oriented
+	 * tasklet
+	 * @return a {@link SimpleStepBuilder}
+	 * @param <I> the type of item to be processed as input
+	 * @param <O> the type of item to be output
+	 * @since 5.0
+	 */
+	public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy,
+			PlatformTransactionManager transactionManager) {
+		return new SimpleStepBuilder<I, O>(this).transactionManager(transactionManager).chunk(completionPolicy);
+	}
+
 	/**
 	 * Create a partition step builder for a remote (or local) step.
 	 * @param stepName the name of the remote or delegate step

spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java

@@ -16,11 +16,13 @@
 package org.springframework.batch.core.step.builder;
 
 import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.transaction.PlatformTransactionManager;
 
 /**
  * Builder for tasklet step based on a custom tasklet (not item oriented).
  *
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  * @since 2.2
  */
 public class TaskletStepBuilder extends AbstractTaskletStepBuilder<TaskletStepBuilder> {
@@ -39,12 +41,26 @@ public TaskletStepBuilder(StepBuilderHelper<?> parent) {
 	/**
 	 * @param tasklet the tasklet to use
 	 * @return this for fluent chaining
+	 * @deprecated use
+	 * {@link TaskletStepBuilder#tasklet(Tasklet, PlatformTransactionManager)}
 	 */
+	@Deprecated(since = "5.0")
 	public TaskletStepBuilder tasklet(Tasklet tasklet) {
 		this.tasklet = tasklet;
 		return this;
 	}
 
+	/**
+	 * @param tasklet the tasklet to use
+	 * @return this for fluent chaining
+	 * @since 5.0
+	 */
+	public TaskletStepBuilder tasklet(Tasklet tasklet, PlatformTransactionManager transactionManager) {
+		this.tasklet = tasklet;
+		super.transactionManager(transactionManager);
+		return this;
+	}
+
 	@Override
 	protected TaskletStepBuilder self() {
 		return this;

spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/FaultTolerantStepFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2013 the original author or authors.
+ * Copyright 2006-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -214,7 +214,7 @@ public void setProcessorTransactional(boolean processorTransactional) {
 
 	@Override
 	protected SimpleStepBuilder<T, S> createBuilder(String name) {
-		return new FaultTolerantStepBuilder<>(new StepBuilder(name));
+		return new FaultTolerantStepBuilder<>(new StepBuilder(name, jobRepository));
 	}
 
 	@Override

spring-batch-core/src/main/java/org/springframework/batch/core/step/factory/SimpleStepFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2013 the original author or authors.
+ * Copyright 2006-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,6 +57,7 @@
  * @see FaultTolerantStepFactoryBean
  * @author Dave Syer
  * @author Robert Kasanicky
+ * @author Mahmoud Ben Hassine
  *
  */
 public class SimpleStepFactoryBean<T, S> implements FactoryBean<Step>, BeanNameAware {
@@ -81,7 +82,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean<Step>, BeanNameA
 
 	private int transactionTimeout = DefaultTransactionAttribute.TIMEOUT_DEFAULT;
 
-	private JobRepository jobRepository;
+	protected JobRepository jobRepository;
 
 	private boolean singleton = true;
 
@@ -318,7 +319,7 @@ public final Step getObject() throws Exception {
 	}
 
 	protected SimpleStepBuilder<T, S> createBuilder(String name) {
-		return new SimpleStepBuilder<>(new StepBuilder(name));
+		return new SimpleStepBuilder<>(new StepBuilder(name, jobRepository));
 	}
 
 	@Override