Fix step annotation listener detection

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

@@ -27,10 +27,12 @@
 import org.springframework.batch.core.annotation.AfterChunk;
 import org.springframework.batch.core.annotation.AfterProcess;
 import org.springframework.batch.core.annotation.AfterRead;
+import org.springframework.batch.core.annotation.AfterStep;
 import org.springframework.batch.core.annotation.AfterWrite;
 import org.springframework.batch.core.annotation.BeforeChunk;
 import org.springframework.batch.core.annotation.BeforeProcess;
 import org.springframework.batch.core.annotation.BeforeRead;
+import org.springframework.batch.core.annotation.BeforeStep;
 import org.springframework.batch.core.annotation.BeforeWrite;
 import org.springframework.batch.core.annotation.OnChunkError;
 import org.springframework.batch.core.annotation.OnProcessError;
@@ -236,6 +238,8 @@ public ChunkOrientedStepBuilder<I, O> listener(StepListener listener) {
 	 */
 	public ChunkOrientedStepBuilder<I, O> listener(Object listener) {
 		Set<Method> listenerMethods = new HashSet<>();
+		listenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), BeforeStep.class));
+		listenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), AfterStep.class));
 		listenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), BeforeChunk.class));
 		listenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), AfterChunk.class));
 		listenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnChunkError.class));

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedStepTests.java

@@ -22,6 +22,8 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.annotation.AfterStep;
+import org.springframework.batch.core.annotation.BeforeStep;
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.job.JobInstance;
 import org.springframework.batch.core.job.parameters.JobParameters;
@@ -51,6 +53,7 @@
  * @author Mahmoud Ben Hassine
  * @author Andrey Litvitski
  * @author xeounxzxu
+ * @author Taeik Lim
  */
 public class ChunkOrientedStepTests {
 
@@ -70,6 +73,39 @@ void testInheritedPropertiesOnBuild() {
 		Assertions.assertEquals(5, step.getStartLimit());
 	}
 
+	@Test
+	void testStepAnnotationBasedListenerRegisteredWithChunkOrientedStepBuilder() throws Exception {
+		AtomicInteger beforeStepCount = new AtomicInteger();
+		AtomicInteger afterStepCount = new AtomicInteger();
+		Object listener = new Object() {
+
+			@BeforeStep
+			public void beforeStep(StepExecution stepExecution) {
+				beforeStepCount.incrementAndGet();
+			}
+
+			@AfterStep
+			public void afterStep(StepExecution stepExecution) {
+				afterStepCount.incrementAndGet();
+			}
+
+		};
+		ChunkOrientedStep<String, String> step = new StepBuilder("step", new ResourcelessJobRepository())
+			.<String, String>chunk(1)
+			.reader(new ListItemReader<>(List.of("item")))
+			.writer(items -> {
+			})
+			.listener(listener)
+			.build();
+		JobExecution jobExecution = new JobExecution(1L, new JobInstance(1L, "job"), new JobParameters());
+		StepExecution stepExecution = new StepExecution(1L, "step", jobExecution);
+
+		step.execute(stepExecution);
+
+		assertEquals(1, beforeStepCount.get());
+		assertEquals(1, afterStepCount.get());
+	}
+
 	@Test
 	void testFaultTolerantChunkOrientedStepSetupWithDefaultSkipLimit() {
 		Assertions.assertDoesNotThrow(() -> new StepBuilder(mock()).chunk(5)