Fix annotation based listener registration in FaultTolerantStepBuilder

Resolves #4137

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.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.
@@ -195,8 +195,7 @@ protected Tasklet createTasklet() {
 	 * @return this for fluent chaining
 	 */
 	@Override
-	@SuppressWarnings("unchecked")
-	public SimpleStepBuilder<I, O> listener(Object listener) {
+	public FaultTolerantStepBuilder<I, O> listener(Object listener) {
 		super.listener(listener);
 
 		Set<Method> skipListenerMethods = new HashSet<>();
@@ -210,9 +209,7 @@ public SimpleStepBuilder<I, O> listener(Object listener) {
 			skipListeners.add((SkipListener) factory.getObject());
 		}
 
-		@SuppressWarnings("unchecked")
-		SimpleStepBuilder<I, O> result = this;
-		return result;
+		return this;
 	}
 
 
@@ -432,7 +429,7 @@ public AbstractTaskletStepBuilder<SimpleStepBuilder<I, O>> stream(ItemStream str
 		}
 		return this;
 	}
-	
+
 	/**
 	 * Override parent method to prevent creation of a new FaultTolerantStepBuilder
 	 */

spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-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.
@@ -15,15 +15,38 @@
  */
 package org.springframework.batch.core.step.builder;
 
+import org.junit.Assert;
 import org.junit.Test;
 
-import static org.junit.Assert.assertEquals;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.configuration.xml.DummyItemReader;
+import org.springframework.batch.core.configuration.xml.DummyItemWriter;
+import org.springframework.batch.core.configuration.xml.DummyJobRepository;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 
+import static org.junit.Assert.assertEquals;
 public class FaultTolerantStepBuilderTests {
 
-    @Test
-    public void faultTolerantReturnsSameInstance() {
-        FaultTolerantStepBuilder<Object, Object> builder = new FaultTolerantStepBuilder<>(new StepBuilder("test"));
-        assertEquals(builder, builder.faultTolerant());
-    }
+	@Test
+	public void faultTolerantReturnsSameInstance() {
+		FaultTolerantStepBuilder<Object, Object> builder = new FaultTolerantStepBuilder<>(new StepBuilder("test"));
+		assertEquals(builder, builder.faultTolerant());
+	}
+
+	@Test
+	public void testAnnotationBasedStepExecutionListenerRegistration() {
+		// given
+		FaultTolerantStepBuilder<Object, Object> faultTolerantStepBuilder = new StepBuilder("myStep")
+				.repository(new DummyJobRepository())
+				.transactionManager(new ResourcelessTransactionManager())
+				.<Object, Object>chunk(5).reader(new DummyItemReader()).writer(new DummyItemWriter()).faultTolerant()
+				.listener(new StepBuilderTests.AnnotationBasedStepExecutionListener());
+
+		// when
+		Step step = faultTolerantStepBuilder.build();
+
+		// then
+		Assert.assertNotNull(step);
+	}
+
 }