diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java
index 1481d9ca6..231e8508d 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2018 the original author or authors.
+ * Copyright 2012-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.
@@ -51,8 +51,7 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
  * @since 2.2
  * @param <B> the type of builder represented
  */
-public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBuilder<B>>
-		extends StepBuilderHelper<AbstractTaskletStepBuilder<B>> {
+public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBuilder<B>> extends StepBuilderHelper<B> {
 
 	protected Set<ChunkListener> chunkListeners = new LinkedHashSet<>();
 
@@ -137,9 +136,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param listener the listener to register
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> listener(ChunkListener listener) {
+	public B listener(ChunkListener listener) {
 		chunkListeners.add(listener);
-		return this;
+		return self();
 	}
 
 	/**
@@ -162,9 +161,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 			this.listener((ChunkListener) factory.getObject());
 		}
 
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	/**
@@ -172,9 +169,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param stream the stream to register
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> stream(ItemStream stream) {
+	public B stream(ItemStream stream) {
 		streams.add(stream);
-		return this;
+		return self();
 	}
 
 	/**
@@ -183,9 +180,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param taskExecutor the task executor to register
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> taskExecutor(TaskExecutor taskExecutor) {
+	public B taskExecutor(TaskExecutor taskExecutor) {
 		this.taskExecutor = taskExecutor;
-		return this;
+		return self();
 	}
 
 	/**
@@ -196,9 +193,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param throttleLimit maximum number of concurrent tasklet executions allowed
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> throttleLimit(int throttleLimit) {
+	public B throttleLimit(int throttleLimit) {
 		this.throttleLimit = throttleLimit;
-		return this;
+		return self();
 	}
 
 	/**
@@ -207,9 +204,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param exceptionHandler the exception handler
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> exceptionHandler(ExceptionHandler exceptionHandler) {
+	public B exceptionHandler(ExceptionHandler exceptionHandler) {
 		this.exceptionHandler = exceptionHandler;
-		return this;
+		return self();
 	}
 
 	/**
@@ -218,9 +215,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param repeatTemplate a repeat template with rules for iterating
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> stepOperations(RepeatOperations repeatTemplate) {
+	public B stepOperations(RepeatOperations repeatTemplate) {
 		this.stepOperations = repeatTemplate;
-		return this;
+		return self();
 	}
 
 	/**
@@ -230,9 +227,9 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @param transactionAttribute a transaction attribute set
 	 * @return this for fluent chaining
 	 */
-	public AbstractTaskletStepBuilder<B> transactionAttribute(TransactionAttribute transactionAttribute) {
+	public B transactionAttribute(TransactionAttribute transactionAttribute) {
 		this.transactionAttribute = transactionAttribute;
-		return this;
+		return self();
 	}
 
 	/**
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
index f6d4a57c9..5e7947c61 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
@@ -224,8 +224,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
 	}
 
 	@Override
-	public AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> transactionAttribute(
-			TransactionAttribute transactionAttribute) {
+	public SimpleStepBuilder<I, O> transactionAttribute(TransactionAttribute transactionAttribute) {
 		return super.transactionAttribute(getTransactionAttribute(transactionAttribute));
 	}
 
@@ -394,7 +393,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
 	}
 
 	@Override
-	public AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> stream(ItemStream stream) {
+	public SimpleStepBuilder<I, O> stream(ItemStream stream) {
 		if (stream instanceof ItemReader<?>) {
 			if (!streamIsReader) {
 				streamIsReader = true;
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FlowStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FlowStepBuilder.java
index 7b7f4a552..d838c9075 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FlowStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FlowStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2011 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.
@@ -69,4 +69,9 @@ public class FlowStepBuilder extends StepBuilderHelper<FlowStepBuilder> {
 		return step;
 	}
 
+	@Override
+	protected FlowStepBuilder self() {
+		return this;
+	}
+
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java
index 13a132561..f9d4609f7 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/JobStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2011 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.
@@ -114,4 +114,9 @@ public class JobStepBuilder extends StepBuilderHelper<JobStepBuilder> {
 
 	}
 
+	@Override
+	protected JobStepBuilder self() {
+		return this;
+	}
+
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java
index 92668b31f..cd4ffa0cb 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2021 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.
@@ -222,6 +222,11 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
 
 	}
 
+	@Override
+	protected PartitionStepBuilder self() {
+		return this;
+	}
+
 	protected TaskExecutor getTaskExecutor() {
 		return taskExecutor;
 	}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
index c1300e1ce..e1d3cc69d 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
@@ -246,7 +246,6 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
 	 * @param listener the object that has a method configured with listener annotation
 	 * @return this for fluent chaining
 	 */
-	@SuppressWarnings("unchecked")
 	@Override
 	public SimpleStepBuilder<I, O> listener(Object listener) {
 		super.listener(listener);
@@ -268,9 +267,7 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
 			itemListeners.add((StepListener) factory.getObject());
 		}
 
-		@SuppressWarnings("unchecked")
-		SimpleStepBuilder<I, O> result = this;
-		return result;
+		return this;
 	}
 
 	/**
@@ -315,6 +312,11 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
 		return this;
 	}
 
+	@Override
+	protected SimpleStepBuilder<I, O> self() {
+		return this;
+	}
+
 	protected RepeatOperations createChunkOperations() {
 		RepeatOperations repeatOperations = chunkOperations;
 		if (repeatOperations == null) {
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java
index e20bd2d6d..8ef8ff959 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilder.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.
@@ -124,4 +124,9 @@ public class StepBuilder extends StepBuilderHelper<StepBuilder> {
 		return new FlowStepBuilder(this).flow(flow);
 	}
 
+	@Override
+	protected StepBuilder self() {
+		return this;
+	}
+
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java
index 135f673f7..a2f63f0f9 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2014 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.
@@ -64,23 +64,17 @@ public abstract class StepBuilderHelper<B extends StepBuilderHelper<B>> {
 
 	public B repository(JobRepository jobRepository) {
 		properties.jobRepository = jobRepository;
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	public B transactionManager(PlatformTransactionManager transactionManager) {
 		properties.transactionManager = transactionManager;
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	public B startLimit(int startLimit) {
 		properties.startLimit = startLimit;
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	/**
@@ -99,25 +93,21 @@ public abstract class StepBuilderHelper<B extends StepBuilderHelper<B>> {
 			properties.addStepExecutionListener((StepExecutionListener) factory.getObject());
 		}
 
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	public B listener(StepExecutionListener listener) {
 		properties.addStepExecutionListener(listener);
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
 	public B allowStartIfComplete(boolean allowStartIfComplete) {
 		properties.allowStartIfComplete = allowStartIfComplete;
-		@SuppressWarnings("unchecked")
-		B result = (B) this;
-		return result;
+		return self();
 	}
 
+	protected abstract B self();
+
 	protected String getName() {
 		return properties.name;
 	}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java
index b4e5aa68c..a37036690 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/TaskletStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2011 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.
@@ -45,6 +45,11 @@ public class TaskletStepBuilder extends AbstractTaskletStepBuilder<TaskletStepBu
 		return this;
 	}
 
+	@Override
+	protected TaskletStepBuilder self() {
+		return this;
+	}
+
 	@Override
 	protected Tasklet createTasklet() {
 		return tasklet;
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java
index b0745657d..aee0a1803 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/StepBuilderTests.java
@@ -17,12 +17,15 @@ package org.springframework.batch.core.step.builder;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.ItemReadListener;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
@@ -40,12 +43,16 @@ import org.springframework.batch.core.annotation.BeforeWrite;
 import org.springframework.batch.core.configuration.xml.DummyItemReader;
 import org.springframework.batch.core.configuration.xml.DummyItemWriter;
 import org.springframework.batch.core.job.SimpleJob;
+import org.springframework.batch.core.listener.ChunkListenerSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
 import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStreamSupport;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.item.support.ListItemWriter;
 import org.springframework.batch.item.support.PassThroughItemProcessor;
+import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
+import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.jdbc.support.JdbcTransactionManager;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
@@ -66,6 +73,10 @@ public class StepBuilderTests {
 
 	private JobRepository jobRepository;
 
+	private StepExecution execution;
+
+	private PlatformTransactionManager transactionManager;
+
 	@Before
 	public void setUp() throws Exception {
 		EmbeddedDatabase embeddedDatabase = new EmbeddedDatabaseBuilder()
@@ -77,14 +88,13 @@ public class StepBuilderTests {
 		factory.setTransactionManager(transactionManager);
 		factory.afterPropertiesSet();
 		this.jobRepository = factory.getObject();
+		this.execution = this.jobRepository.createJobExecution("foo", new JobParameters()).createStepExecution("step");
+		this.jobRepository.add(this.execution);
+		this.transactionManager = new ResourcelessTransactionManager();
 	}
 
 	@Test
 	public void test() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		TaskletStepBuilder builder = new StepBuilder("step").repository(jobRepository)
 				.transactionManager(transactionManager).tasklet((contribution, chunkContext) -> null);
 		builder.build().execute(execution);
@@ -93,10 +103,6 @@ public class StepBuilderTests {
 
 	@Test
 	public void testListeners() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		TaskletStepBuilder builder = new StepBuilder("step").repository(jobRepository)
 				.transactionManager(transactionManager).listener(new InterfaceBasedStepExecutionListener())
 				.listener(new AnnotationBasedStepExecutionListener()).tasklet((contribution, chunkContext) -> null);
@@ -112,10 +118,6 @@ public class StepBuilderTests {
 
 	@Test
 	public void testAnnotationBasedChunkListenerForTaskletStep() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		TaskletStepBuilder builder = new StepBuilder("step").repository(jobRepository)
 				.transactionManager(transactionManager).tasklet((contribution, chunkContext) -> null)
 				.listener(new AnnotationBasedChunkListener());
@@ -127,10 +129,6 @@ public class StepBuilderTests {
 
 	@Test
 	public void testAnnotationBasedChunkListenerForSimpleTaskletStep() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		SimpleStepBuilder<Object, Object> builder = new StepBuilder("step").repository(jobRepository)
 				.transactionManager(transactionManager).chunk(5).reader(new DummyItemReader())
 				.writer(new DummyItemWriter()).listener(new AnnotationBasedChunkListener());
@@ -142,13 +140,9 @@ public class StepBuilderTests {
 
 	@Test
 	public void testAnnotationBasedChunkListenerForFaultTolerantTaskletStep() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		SimpleStepBuilder<Object, Object> builder = new StepBuilder("step").repository(jobRepository)
 				.transactionManager(transactionManager).chunk(5).reader(new DummyItemReader())
-				.writer(new DummyItemWriter()).faultTolerant().listener(new AnnotationBasedChunkListener()); // TODO
+				.writer(new DummyItemWriter()).faultTolerant().listener(new AnnotationBasedChunkListener()); // TODO//
 																												// should
 																												// this
 																												// return
@@ -161,10 +155,6 @@ public class StepBuilderTests {
 
 	@Test
 	public void testAnnotationBasedChunkListenerForJobStepBuilder() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
 		SimpleJob job = new SimpleJob("job");
 		job.setJobRepository(jobRepository);
 		JobStepBuilder builder = new StepBuilder("step").repository(jobRepository)
@@ -180,11 +170,6 @@ public class StepBuilderTests {
 
 	@Test
 	public void testItemListeners() throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
-
 		List<String> items = Arrays.asList("1", "2", "3");
 
 		ItemReader<String> reader = new ListItemReader<>(items);
@@ -219,11 +204,6 @@ public class StepBuilderTests {
 	}
 
 	private void assertStepFunctions(boolean faultTolerantStep) throws Exception {
-		StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters())
-				.createStepExecution("step");
-		jobRepository.add(execution);
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
-
 		List<Long> items = Arrays.asList(1L, 2L, 3L);
 
 		ItemReader<Long> reader = new ListItemReader<>(items);
@@ -246,11 +226,91 @@ public class StepBuilderTests {
 		assertEquals("3", writtenItems.get(2));
 	}
 
+	@Test
+	public void testReturnedTypeOfChunkListenerIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.listener(new ChunkListenerSupport() {
+		}));
+	}
+
+	@Test
+	public void testReturnedTypeOfStreamIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.stream(new ItemStreamSupport() {
+		}));
+	}
+
+	@Test
+	public void testReturnedTypeOfTaskExecutorIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.taskExecutor(null));
+	}
+
+	@Test
+	public void testReturnedTypeOfThrottleLimitIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.throttleLimit(4));
+	}
+
+	@Test
+	public void testReturnedTypeOfExceptionHandlerIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(
+				builder -> builder.exceptionHandler(new DefaultExceptionHandler()));
+	}
+
+	@Test
+	public void testReturnedTypeOfStepOperationsIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(
+				builder -> builder.stepOperations(new RepeatTemplate()));
+	}
+
+	@Test
+	public void testReturnedTypeOfTransactionAttributeIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.transactionAttribute(null));
+	}
+
+	@Test
+	public void testReturnedTypeOfListenerIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(
+				builder -> builder.listener(new AnnotationBasedStepExecutionListener()));
+		assertEquals(1, AnnotationBasedStepExecutionListener.beforeStepCount);
+		assertEquals(1, AnnotationBasedStepExecutionListener.afterStepCount);
+	}
+
+	@Test
+	public void testReturnedTypeOfExecutionListenerIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(
+				builder -> builder.listener(new InterfaceBasedStepExecutionListener()));
+		assertEquals(1, InterfaceBasedStepExecutionListener.beforeStepCount);
+		assertEquals(1, InterfaceBasedStepExecutionListener.afterStepCount);
+	}
+
+	@Test
+	public void testReturnedTypeOfAllowStartIfCompleteIsAssignableToSimpleStepBuilder() throws Exception {
+		testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(builder -> builder.allowStartIfComplete(false));
+	}
+
+	private void testReturnedTypeOfSetterIsAssignableToSimpleStepBuilder(
+			UnaryOperator<SimpleStepBuilder<String, String>> configurer) throws Exception {
+		List<String> items = Arrays.asList("1", "2", "3");
+		ItemReader<String> reader = new ListItemReader<>(items);
+
+		SimpleStepBuilder<String, String> builder = new StepBuilder("step").repository(jobRepository)
+				.transactionManager(transactionManager).<String, String>chunk(3).reader(reader)
+				.writer(new DummyItemWriter());
+		configurer.apply(builder).listener(new InterfaceBasedItemReadListenerListener()).build().execute(execution);
+
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(4, InterfaceBasedItemReadListenerListener.beforeReadCount);
+		assertEquals(3, InterfaceBasedItemReadListenerListener.afterReadCount);
+	}
+
 	public static class InterfaceBasedStepExecutionListener implements StepExecutionListener {
 
 		static int beforeStepCount = 0;
 		static int afterStepCount = 0;
 
+		public InterfaceBasedStepExecutionListener() {
+			beforeStepCount = 0;
+			afterStepCount = 0;
+		}
+
 		@Override
 		public void beforeStep(StepExecution stepExecution) {
 			beforeStepCount++;
@@ -265,6 +325,32 @@ public class StepBuilderTests {
 
 	}
 
+	public static class InterfaceBasedItemReadListenerListener implements ItemReadListener<String> {
+
+		static int beforeReadCount = 0;
+		static int afterReadCount = 0;
+
+		public InterfaceBasedItemReadListenerListener() {
+			beforeReadCount = 0;
+			afterReadCount = 0;
+		}
+
+		@Override
+		public void beforeRead() {
+			beforeReadCount++;
+		}
+
+		@Override
+		public void afterRead(String item) {
+			afterReadCount++;
+		}
+
+		@Override
+		public void onReadError(Exception ex) {
+		}
+
+	}
+
 	@SuppressWarnings("unused")
 	public static class AnnotationBasedStepExecutionListener {