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

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-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.
@@ -85,11 +85,23 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
 	/**
 	 * Create a new {@link RemoteChunkingManagerStepBuilder}.
 	 * @param stepName name of the manager step
+	 * @deprecated use {@link }
 	 */
+	@Deprecated(since = "5.0")
 	public RemoteChunkingManagerStepBuilder(String stepName) {
 		super(new StepBuilder(stepName));
 	}
 
+	/**
+	 * Create a new {@link RemoteChunkingManagerStepBuilder}.
+	 * @param stepName name of the manager step
+	 * @param jobRepository the job repository the step should report to
+	 * @since 5.0
+	 */
+	public RemoteChunkingManagerStepBuilder(String stepName, JobRepository jobRepository) {
+		super(new StepBuilder(stepName, jobRepository));
+	}
+
 	/**
 	 * Set the input channel on which replies from workers will be received. The provided
 	 * input channel will be set as a reply channel on the

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-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.
@@ -52,7 +52,7 @@ public class RemoteChunkingManagerStepBuilderFactory {
 	 * @return a {@link RemoteChunkingManagerStepBuilder}
 	 */
 	public <I, O> RemoteChunkingManagerStepBuilder<I, O> get(String name) {
-		return new RemoteChunkingManagerStepBuilder<I, O>(name).repository(this.jobRepository)
+		return new RemoteChunkingManagerStepBuilder<I, O>(name, this.jobRepository)
 				.transactionManager(this.transactionManager);
 	}
 

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilder.java

@@ -82,11 +82,24 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
 	/**
 	 * Create a new {@link RemotePartitioningManagerStepBuilder}.
 	 * @param stepName name of the manager step
+	 * @deprecated use
+	 * {@link RemotePartitioningManagerStepBuilder#RemotePartitioningManagerStepBuilder(String, JobRepository)}
 	 */
+	@Deprecated
 	public RemotePartitioningManagerStepBuilder(String stepName) {
 		super(new StepBuilder(stepName));
 	}
 
+	/**
+	 * Create a new {@link RemotePartitioningManagerStepBuilder}.
+	 * @param stepName name of the manager step
+	 * @param jobRepository job repository to which the step should report to
+	 * @since 5.0
+	 */
+	public RemotePartitioningManagerStepBuilder(String stepName, JobRepository jobRepository) {
+		super(new StepBuilder(stepName, jobRepository));
+	}
+
 	/**
 	 * Set the input channel on which replies from workers will be received.
 	 * @param inputChannel the input channel

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderFactory.java

@@ -62,8 +62,8 @@ public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryA
 	 * @return a {@link RemotePartitioningManagerStepBuilder}
 	 */
 	public RemotePartitioningManagerStepBuilder get(String name) {
-		return new RemotePartitioningManagerStepBuilder(name).repository(this.jobRepository)
-				.jobExplorer(this.jobExplorer).beanFactory(this.beanFactory);
+		return new RemotePartitioningManagerStepBuilder(name, this.jobRepository).jobExplorer(this.jobExplorer)
+				.beanFactory(this.beanFactory);
 	}
 
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java

@@ -83,11 +83,24 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
 	/**
 	 * Initialize a step builder for a step with the given name.
 	 * @param name the name of the step
+	 * @deprecated use
+	 * {@link RemotePartitioningWorkerStepBuilder#RemotePartitioningWorkerStepBuilder(String, JobRepository)}
 	 */
+	@Deprecated(since = "5.0")
 	public RemotePartitioningWorkerStepBuilder(String name) {
 		super(name);
 	}
 
+	/**
+	 * Initialize a step builder for a step with the given name.
+	 * @param name the name of the step
+	 * @param jobRepository the job repository to which the step should report to
+	 * @since 5.0
+	 */
+	public RemotePartitioningWorkerStepBuilder(String name, JobRepository jobRepository) {
+		super(name, jobRepository);
+	}
+
 	/**
 	 * Set the input channel on which step execution requests sent by the manager are
 	 * received.
@@ -174,24 +187,46 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
 		return this;
 	}
 
+	@Deprecated(since = "5.0")
 	@Override
 	public TaskletStepBuilder tasklet(Tasklet tasklet) {
 		configureWorkerIntegrationFlow();
 		return super.tasklet(tasklet);
 	}
 
+	@Override
+	public TaskletStepBuilder tasklet(Tasklet tasklet, PlatformTransactionManager transactionManager) {
+		configureWorkerIntegrationFlow();
+		return super.tasklet(tasklet, transactionManager);
+	}
+
+	@Deprecated(since = "5.0")
 	@Override
 	public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
 		configureWorkerIntegrationFlow();
 		return super.chunk(chunkSize);
 	}
 
+	@Override
+	public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize, PlatformTransactionManager transactionManager) {
+		configureWorkerIntegrationFlow();
+		return super.chunk(chunkSize, transactionManager);
+	}
+
+	@Deprecated(since = "5.0")
 	@Override
 	public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
 		configureWorkerIntegrationFlow();
 		return super.chunk(completionPolicy);
 	}
 
+	@Override
+	public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy,
+			PlatformTransactionManager transactionManager) {
+		configureWorkerIntegrationFlow();
+		return super.chunk(completionPolicy, transactionManager);
+	}
+
 	@Override
 	public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) {
 		configureWorkerIntegrationFlow();

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java

@@ -62,8 +62,8 @@ public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAw
 	 * @return a {@link RemotePartitioningWorkerStepBuilder}
 	 */
 	public RemotePartitioningWorkerStepBuilder get(String name) {
-		return new RemotePartitioningWorkerStepBuilder(name).repository(this.jobRepository)
-				.jobExplorer(this.jobExplorer).beanFactory(this.beanFactory);
+		return new RemotePartitioningWorkerStepBuilder(name, this.jobRepository).jobExplorer(this.jobExplorer)
+				.beanFactory(this.beanFactory);
 	}
 
 }

spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderTests.java

@@ -96,7 +96,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void inputChannelMustNotBeNull() {
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").inputChannel(null).build());
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.inputChannel(null).build());
 
 		// then
 		assertThat(expectedException).hasMessage("inputChannel must not be null");
@@ -106,7 +107,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void outputChannelMustNotBeNull() {
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").outputChannel(null).build());
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.outputChannel(null).build());
 
 		// then
 		assertThat(expectedException).hasMessage("outputChannel must not be null");
@@ -116,7 +118,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void messagingTemplateMustNotBeNull() {
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").messagingTemplate(null).build());
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.messagingTemplate(null).build());
 
 		// then
 		assertThat(expectedException).hasMessage("messagingTemplate must not be null");
@@ -126,7 +129,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void maxWaitTimeoutsMustBeGreaterThanZero() {
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").maxWaitTimeouts(-1).build());
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.maxWaitTimeouts(-1).build());
 
 		// then
 		assertThat(expectedException).hasMessage("maxWaitTimeouts must be greater than zero");
@@ -136,7 +140,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void throttleLimitMustNotBeGreaterThanZero() {
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").throttleLimit(-1L).build());
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.throttleLimit(-1L).build());
 
 		// then
 		assertThat(expectedException).hasMessage("throttleLimit must be greater than zero");
@@ -145,7 +150,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	@Test
 	void testMandatoryInputChannel() {
 		// given
-		RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<>("step");
+		RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<>("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class, builder::build);
@@ -158,7 +164,7 @@ class RemoteChunkingManagerStepBuilderTests {
 	void eitherOutputChannelOrMessagingTemplateMustBeProvided() {
 		// given
 		RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<String, String>(
-				"step").inputChannel(this.inputChannel).outputChannel(new DirectChannel())
+				"step", this.jobRepository).inputChannel(this.inputChannel).outputChannel(new DirectChannel())
 						.messagingTemplate(new MessagingTemplate());
 
 		// when
@@ -173,8 +179,8 @@ class RemoteChunkingManagerStepBuilderTests {
 	void testUnsupportedOperationExceptionWhenSpecifyingAnItemWriter() {
 		// when
 		final Exception expectedException = assertThrows(UnsupportedOperationException.class,
-				() -> new RemoteChunkingManagerStepBuilder<String, String>("step").reader(this.itemReader)
-						.writer(items -> {
+				() -> new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+						.reader(this.itemReader).writer(items -> {
 						}).repository(this.jobRepository).transactionManager(this.transactionManager)
 						.inputChannel(this.inputChannel).outputChannel(this.outputChannel).build());
 
@@ -188,9 +194,9 @@ class RemoteChunkingManagerStepBuilderTests {
 	@Test
 	void testManagerStepCreation() {
 		// when
-		TaskletStep taskletStep = new RemoteChunkingManagerStepBuilder<String, String>("step").reader(this.itemReader)
-				.repository(this.jobRepository).transactionManager(this.transactionManager)
-				.inputChannel(this.inputChannel).outputChannel(this.outputChannel).build();
+		TaskletStep taskletStep = new RemoteChunkingManagerStepBuilder<String, String>("step", this.jobRepository)
+				.reader(this.itemReader).transactionManager(this.transactionManager).inputChannel(this.inputChannel)
+				.outputChannel(this.outputChannel).build();
 
 		// then
 		assertNotNull(taskletStep);

spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderTests.java

@@ -55,7 +55,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void inputChannelMustNotBeNull() {
 		// given
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -68,7 +69,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void outputChannelMustNotBeNull() {
 		// given
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -81,7 +83,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void messagingTemplateMustNotBeNull() {
 		// given
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -94,7 +97,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void jobExplorerMustNotBeNull() {
 		// given
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -107,7 +111,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void pollIntervalMustBeGreaterThanZero() {
 		// given
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -120,8 +125,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	@Test
 	void eitherOutputChannelOrMessagingTemplateMustBeProvided() {
 		// given
-		RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step")
-				.outputChannel(new DirectChannel()).messagingTemplate(new MessagingTemplate());
+		RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository).outputChannel(new DirectChannel()).messagingTemplate(new MessagingTemplate());
 
 		// when
 		final Exception expectedException = assertThrows(IllegalStateException.class, builder::build);
@@ -135,7 +140,8 @@ class RemotePartitioningManagerStepBuilderTests {
 	void testUnsupportedOperationExceptionWhenSpecifyingPartitionHandler() {
 		// given
 		PartitionHandler partitionHandler = Mockito.mock(PartitionHandler.class);
-		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step");
+		final RemotePartitioningManagerStepBuilder builder = new RemotePartitioningManagerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(UnsupportedOperationException.class,
@@ -161,7 +167,7 @@ class RemotePartitioningManagerStepBuilderTests {
 		};
 
 		// when
-		Step step = new RemotePartitioningManagerStepBuilder("managerStep").repository(jobRepository)
+		Step step = new RemotePartitioningManagerStepBuilder("managerStep", this.jobRepository)
 				.outputChannel(outputChannel).partitioner("workerStep", partitioner).gridSize(gridSize)
 				.pollInterval(pollInterval).timeout(timeout).startLimit(startLimit).aggregator(stepExecutionAggregator)
 				.allowStartIfComplete(true).build();
@@ -198,7 +204,7 @@ class RemotePartitioningManagerStepBuilderTests {
 		};
 
 		// when
-		Step step = new RemotePartitioningManagerStepBuilder("managerStep").repository(jobRepository)
+		Step step = new RemotePartitioningManagerStepBuilder("managerStep", this.jobRepository)
 				.outputChannel(outputChannel).partitioner("workerStep", partitioner).gridSize(gridSize)
 				.startLimit(startLimit).aggregator(stepExecutionAggregator).allowStartIfComplete(true).build();
 

spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderTests.java

@@ -19,8 +19,10 @@ package org.springframework.batch.integration.partition;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mock;
 
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.transaction.PlatformTransactionManager;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -33,10 +35,17 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Mock
 	private Tasklet tasklet;
 
+	@Mock
+	private JobRepository jobRepository;
+
+	@Mock
+	private PlatformTransactionManager transactionManager;
+
 	@Test
 	void inputChannelMustNotBeNull() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -49,7 +58,8 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Test
 	void outputChannelMustNotBeNull() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -62,7 +72,8 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Test
 	void jobExplorerMustNotBeNull() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -75,7 +86,8 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Test
 	void stepLocatorMustNotBeNull() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -88,7 +100,8 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Test
 	void beanFactoryMustNotBeNull() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
@@ -101,11 +114,12 @@ class RemotePartitioningWorkerStepBuilderTests {
 	@Test
 	void testMandatoryInputChannel() {
 		// given
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step");
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> builder.tasklet(this.tasklet));
+				() -> builder.tasklet(this.tasklet, this.transactionManager));
 
 		// then
 		assertThat(expectedException).hasMessage("An InputChannel must be provided");
@@ -115,12 +129,12 @@ class RemotePartitioningWorkerStepBuilderTests {
 	void testMandatoryJobExplorer() {
 		// given
 		DirectChannel inputChannel = new DirectChannel();
-		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step")
-				.inputChannel(inputChannel);
+		final RemotePartitioningWorkerStepBuilder builder = new RemotePartitioningWorkerStepBuilder("step",
+				this.jobRepository).inputChannel(inputChannel);
 
 		// when
 		final Exception expectedException = assertThrows(IllegalArgumentException.class,
-				() -> builder.tasklet(this.tasklet));
+				() -> builder.tasklet(this.tasklet, this.transactionManager));
 
 		// then
 		assertThat(expectedException).hasMessage("A JobExplorer must be provided");