Add AbstractTaskletStepBuilder copy constructor

This commit includes tests for the copy constructor
of AbstractTaskletStepBuilder and for the faultTolerant
method, specifically after taskExecutor has been set.

(cherry picked from commit b9ba8ffb86)

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

@@ -49,6 +49,7 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
  * @author Dave Syer
  * @author Michael Minella
  * @author Mahmoud Ben Hassine
+ * @author Ilpyo Yang
  * @since 2.2
  * @param <B> the type of builder represented
  */
@@ -74,6 +75,23 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 		super(parent);
 	}
 
+	/**
+	 * Create a new builder initialized with any properties in the parent. The parent is
+	 * copied, so it can be re-used.
+	 * @param parent a parent helper containing common step properties
+	 */
+	public AbstractTaskletStepBuilder(AbstractTaskletStepBuilder<?> parent) {
+		super(parent);
+		this.chunkListeners = parent.chunkListeners;
+		this.stepOperations = parent.stepOperations;
+		this.transactionManager = parent.transactionManager;
+		this.transactionAttribute = parent.transactionAttribute;
+		this.streams.addAll(parent.streams);
+		this.exceptionHandler = parent.exceptionHandler;
+		this.throttleLimit = parent.throttleLimit;
+		this.taskExecutor = parent.taskExecutor;
+	}
+
 	protected abstract Tasklet createTasklet();
 
 	/**

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

@@ -114,7 +114,6 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
 		this.itemListeners = parent.itemListeners;
 		this.readerTransactionalQueue = parent.readerTransactionalQueue;
 		this.meterRegistry = parent.meterRegistry;
-		this.transactionManager(parent.getTransactionManager());
 	}
 
 	public FaultTolerantStepBuilder<I, O> faultTolerant() {

spring-batch-test/src/test/java/org/springframework/batch/test/AbstractTaskletStepBuilderTests.java

@@ -0,0 +1,114 @@
+package org.springframework.batch.test;
+/*
+ * Copyright 2020-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.
+ * You may obtain a copy of the License at
+ *
+ *       https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.AbstractTaskletStepBuilder;
+import org.springframework.batch.core.step.builder.SimpleStepBuilder;
+import org.springframework.batch.core.step.builder.StepBuilderHelper;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.test.context.SpringBatchTest;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * Test cases for verifying the {@link AbstractTaskletStepBuilder} and faultTolerant() functionality.
+ *
+ * @author Ilpyo Yang
+ */
+@SpringBatchTest
+@SpringJUnitConfig
+public class AbstractTaskletStepBuilderTests {
+	private final JobRepository jobRepository =  mock(JobRepository.class);
+	private final int chunkSize = 10;
+	private final ItemReader itemReader = mock(ItemReader.class);
+	private final ItemProcessor itemProcessor = mock(ItemProcessor.class);
+	private final ItemWriter itemWriter = mock(ItemWriter.class);
+	private final SimpleAsyncTaskExecutor taskExecutor  = new SimpleAsyncTaskExecutor();
+	SimpleStepBuilder simpleStepBuilder;
+
+	private <T> T accessPrivateField(Object o, String fieldName) throws ReflectiveOperationException {
+		Field field = o.getClass().getDeclaredField(fieldName);
+		field.setAccessible(true);
+		return (T) field.get(o);
+	}
+
+	private <T> T accessSuperClassPrivateField(Object o, String fieldName) throws ReflectiveOperationException {
+		Field field = o.getClass().getSuperclass().getDeclaredField(fieldName);
+		field.setAccessible(true);
+		return (T) field.get(o);
+	}
+
+	@BeforeEach
+	void set(){
+		StepBuilderHelper stepBuilderHelper = new StepBuilderHelper("test", jobRepository) {
+			@Override
+			protected StepBuilderHelper self() {
+				return null;
+			}
+		};
+		simpleStepBuilder = new SimpleStepBuilder(stepBuilderHelper);
+		simpleStepBuilder.chunk(chunkSize);
+		simpleStepBuilder.reader(itemReader);
+		simpleStepBuilder.processor(itemProcessor);
+		simpleStepBuilder.writer(itemWriter);
+	}
+
+	@Test
+	void copyConstractorTest() throws ReflectiveOperationException {
+		Constructor<SimpleStepBuilder> constructor = SimpleStepBuilder.class.getDeclaredConstructor(SimpleStepBuilder.class);
+		constructor.setAccessible(true);
+		SimpleStepBuilder copySimpleStepBuilder = constructor.newInstance(simpleStepBuilder);
+
+		int copyChunkSize = accessPrivateField(copySimpleStepBuilder, "chunkSize");
+		ItemReader copyItemReader = accessPrivateField(copySimpleStepBuilder, "reader");
+		ItemProcessor copyItemProcessor = accessPrivateField(copySimpleStepBuilder, "processor");
+		ItemWriter copyItemWriter = accessPrivateField(copySimpleStepBuilder, "writer");
+
+		assertEquals(chunkSize, copyChunkSize);
+		assertEquals(itemReader, copyItemReader);
+		assertEquals(itemProcessor, copyItemProcessor);
+		assertEquals(itemWriter, copyItemWriter);
+	}
+
+	@Test
+	void faultTolerantMethodTest() throws ReflectiveOperationException {
+		simpleStepBuilder.taskExecutor(taskExecutor); // The task executor is set before faultTolerant()
+		simpleStepBuilder.faultTolerant();
+
+		int afterChunkSize = accessPrivateField(simpleStepBuilder, "chunkSize");
+		ItemReader afterItemReader = accessPrivateField(simpleStepBuilder, "reader");
+		ItemProcessor afterItemProcessor = accessPrivateField(simpleStepBuilder, "processor");
+		ItemWriter afterItemWriter = accessPrivateField(simpleStepBuilder, "writer");
+		TaskExecutor afterTaskExecutor = accessSuperClassPrivateField(simpleStepBuilder, "taskExecutor");
+
+		assertEquals(chunkSize, afterChunkSize);
+		assertEquals(itemReader, afterItemReader);
+		assertEquals(itemProcessor, afterItemProcessor);
+		assertEquals(itemWriter, afterItemWriter);
+		assertEquals(taskExecutor, afterTaskExecutor);
+	}
+}