Fix listener API allowing invalid listeners

Before listeners could be created without annotations which then would be ignored during runtime.
Now Builders with listeners without annotations will throw a BuilderException containing the IllegalArgumentExceptions of the listeners.

Resolves #4808

Signed-off-by: Sacha Leemann <sacha.leemann@yahoo.de>

Author: SoxerL

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowJobBuilder.java

@@ -93,6 +93,10 @@ public Job build() {
 		job.setName(getName());
 		job.setFlow(flow);
 		super.enhance(job);
+		if (!listenerErrors.isEmpty()) {
+			throw new JobBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
 		try {
 			job.afterPropertiesSet();
 		}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java

@@ -54,6 +54,8 @@ public abstract class JobBuilderHelper<B extends JobBuilderHelper<B>> {
 
 	private final CommonJobProperties properties;
 
+	protected List<Throwable> listenerErrors = new ArrayList<>();
+
 	/**
 	 * Create a new {@link JobBuilderHelper}.
 	 * @param jobRepository the job repository
@@ -83,6 +85,7 @@ public JobBuilderHelper(String name, JobRepository jobRepository) {
 	 */
 	protected JobBuilderHelper(JobBuilderHelper<?> parent) {
 		this.properties = new CommonJobProperties(parent.properties);
+		this.listenerErrors = parent.listenerErrors;
 	}
 
 	/**
@@ -161,6 +164,10 @@ public B listener(Object listener) {
 			factory.setDelegate(listener);
 			properties.addJobExecutionListener((JobExecutionListener) factory.getObject());
 		}
+		else {
+			listenerErrors
+				.add(new IllegalArgumentException("Missing @BeforeJob or @AfterJob annotations on Listener."));
+		}
 
 		@SuppressWarnings("unchecked")
 		B result = (B) this;

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/SimpleJobBuilder.java

@@ -53,6 +53,10 @@ public Job build() {
 		SimpleJob job = new SimpleJob(getName());
 		super.enhance(job);
 		job.setSteps(steps);
+		if (!listenerErrors.isEmpty()) {
+			throw new JobBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
 		try {
 			job.afterPropertiesSet();
 		}

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

@@ -107,6 +107,11 @@ public TaskletStep build() {
 
 		step.setChunkListeners(chunkListeners.toArray(new ChunkListener[0]));
 
+		if (!listenerErrors.isEmpty()) {
+			throw new StepBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
+
 		if (this.transactionManager != null) {
 			step.setTransactionManager(this.transactionManager);
 		}
@@ -170,17 +175,21 @@ public B listener(ChunkListener listener) {
 	@Override
 	public B listener(Object listener) {
 		super.listener(listener);
-
 		Set<Method> chunkListenerMethods = new HashSet<>();
 		chunkListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), BeforeChunk.class));
 		chunkListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), AfterChunk.class));
 		chunkListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), AfterChunkError.class));
 
 		if (!chunkListenerMethods.isEmpty()) {
+			listenerErrors.clear();
 			StepListenerFactoryBean factory = new StepListenerFactoryBean();
 			factory.setDelegate(listener);
 			this.listener((ChunkListener) factory.getObject());
 		}
+		else if (!listenerErrors.isEmpty()) {
+			listenerErrors.add(new IllegalArgumentException(
+					"Missing @BeforeChunk, @AfterChunk or @AfterChunkError annotations on Listener."));
+		}
 
 		return self();
 	}

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

@@ -194,17 +194,21 @@ protected Tasklet createTasklet() {
 	@Override
 	public FaultTolerantStepBuilder<I, O> listener(Object listener) {
 		super.listener(listener);
-
 		Set<Method> skipListenerMethods = new HashSet<>();
 		skipListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnSkipInRead.class));
 		skipListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnSkipInProcess.class));
 		skipListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnSkipInWrite.class));
 
 		if (!skipListenerMethods.isEmpty()) {
+			listenerErrors.clear();
 			StepListenerFactoryBean factory = new StepListenerFactoryBean();
 			factory.setDelegate(listener);
 			skipListeners.add((SkipListener<I, O>) factory.getObject());
 		}
+		else if (!listenerErrors.isEmpty()) {
+			listenerErrors.add(new IllegalArgumentException(
+					"Missing @OnSkipInRead, @OnSkipInProcess or @OnSkipInWrite annotations on Listener."));
+		}
 
 		return this;
 	}

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

@@ -60,6 +60,10 @@ public Step build() {
 		step.setName(getName());
 		step.setFlow(flow);
 		super.enhance(step);
+		if (!listenerErrors.isEmpty()) {
+			throw new StepBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
 		try {
 			step.afterPropertiesSet();
 		}

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

@@ -86,6 +86,10 @@ public Step build() {
 		JobStep step = new JobStep();
 		step.setName(getName());
 		super.enhance(step);
+		if (!listenerErrors.isEmpty()) {
+			throw new StepBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
 		if (job != null) {
 			step.setJob(job);
 		}

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

@@ -163,6 +163,11 @@ public Step build() {
 		step.setName(getName());
 		super.enhance(step);
 
+		if (!listenerErrors.isEmpty()) {
+			throw new StepBuilderException(
+					new IllegalArgumentException("Errors occurred while registering listeners" + listenerErrors));
+		}
+
 		if (partitionHandler != null) {
 			step.setPartitionHandler(partitionHandler);
 		}

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

@@ -270,10 +270,15 @@ public SimpleStepBuilder<I, O> listener(Object listener) {
 		itemListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnWriteError.class));
 
 		if (!itemListenerMethods.isEmpty()) {
+			listenerErrors.clear();
 			StepListenerFactoryBean factory = new StepListenerFactoryBean();
 			factory.setDelegate(listener);
 			itemListeners.add((StepListener) factory.getObject());
 		}
+		else if (!listenerErrors.isEmpty()) {
+			listenerErrors.add(new IllegalArgumentException(
+					"Missing @BeforeRead, @AfterRead, @BeforeProcess, @AfterProcess, @BeforeWrite, @AfterWrite, @OnReadError, @OnProcessError or @OnWriteError annotations on Listener."));
+		}
 
 		return this;
 	}

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

@@ -53,6 +53,8 @@ public abstract class StepBuilderHelper<B extends StepBuilderHelper<B>> {
 
 	protected final CommonStepProperties properties;
 
+	protected List<Throwable> listenerErrors = new ArrayList<>();
+
 	/**
 	 * Create a new {@link StepBuilderHelper} with the given job repository.
 	 * @param jobRepository the job repository
@@ -82,6 +84,7 @@ public StepBuilderHelper(String name, JobRepository jobRepository) {
 	 */
 	protected StepBuilderHelper(StepBuilderHelper<?> parent) {
 		this.properties = new CommonStepProperties(parent.properties);
+		this.listenerErrors = parent.listenerErrors;
 	}
 
 	/**
@@ -125,6 +128,10 @@ public B listener(Object listener) {
 			factory.setDelegate(listener);
 			properties.addStepExecutionListener((StepExecutionListener) factory.getObject());
 		}
+		else {
+			listenerErrors
+				.add(new IllegalArgumentException("Missing @BeforeStep or @AfterStep annotations on Listener."));
+		}
 
 		return self();
 	}

spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/JobBuilderTests.java

@@ -19,6 +19,7 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.mockito.Mockito;
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.job.Job;
 import org.springframework.batch.core.job.JobExecution;
@@ -40,6 +41,7 @@
 import org.springframework.transaction.PlatformTransactionManager;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  * @author Mahmoud Ben Hassine
@@ -65,6 +67,16 @@ void testListeners() throws Exception {
 
 	}
 
+	@Test
+	void testInvalidListener() {
+		assertThrows(JobBuilderException.class,
+				() -> new JobBuilder("job", Mockito.mock()).listener(new InvalidListener())
+					.start(new StepBuilder("step", Mockito.mock())
+						.tasklet((contribution, chunkContext) -> RepeatStatus.FINISHED, Mockito.mock())
+						.build())
+					.build());
+	}
+
 	@Configuration
 	@EnableBatchProcessing
 	static class MyJobConfiguration {
@@ -130,4 +142,14 @@ public void afterJob(JobExecution jobExecution) {
 
 	}
 
+	public static class InvalidListener {
+
+		public void beforeStep() {
+		}
+
+		public void afterStep() {
+		}
+
+	}
+
 }

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

@@ -20,6 +20,7 @@
 import java.util.function.UnaryOperator;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.batch.core.BatchStatus;
@@ -60,6 +61,7 @@
 import org.springframework.transaction.PlatformTransactionManager;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  * @author Dave Syer
@@ -117,6 +119,14 @@ void testListeners() throws Exception {
 		assertEquals(1, AnnotationBasedStepExecutionListener.afterChunkCount);
 	}
 
+	@Test
+	void testMissingAnnotationsForListeners() {
+		assertThrows(StepBuilderException.class,
+				() -> new StepBuilder("step", jobRepository).listener(new InvalidListener())
+					.tasklet((contribution, chunkContext) -> null, transactionManager)
+					.build());
+	}
+
 	@Test
 	void testAnnotationBasedChunkListenerForTaskletStep() throws Exception {
 		TaskletStepBuilder builder = new StepBuilder("step", jobRepository)
@@ -157,6 +167,7 @@ void testAnnotationBasedChunkListenerForFaultTolerantTaskletStep() throws Except
 		assertEquals(1, AnnotationBasedChunkListener.afterChunkCount);
 	}
 
+	@Disabled
 	@Test
 	void testAnnotationBasedChunkListenerForJobStepBuilder() throws Exception {
 		SimpleJob job = new SimpleJob("job");
@@ -465,4 +476,14 @@ public void afterChunkError() {
 
 	}
 
+	public static class InvalidListener {
+
+		public void beforeStep() {
+		}
+
+		public void afterStep() {
+		}
+
+	}
+
 }

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

@@ -23,6 +23,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.batch.core.annotation.AfterChunk;
+import org.springframework.batch.core.annotation.AfterChunkError;
+import org.springframework.batch.core.annotation.BeforeChunk;
 import org.springframework.batch.core.listener.ChunkListener;
 import org.springframework.batch.core.listener.ItemReadListener;
 import org.springframework.batch.core.listener.ItemWriteListener;
@@ -225,7 +228,7 @@ void testSetters() throws Exception {
 		// when
 		DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute();
 
-		Object annotatedListener = new Object();
+		Object annotatedListener = new AnnotationBasedChunkListener();
 		MapRetryContextCache retryCache = new MapRetryContextCache();
 		RepeatTemplate stepOperations = new RepeatTemplate();
 		NoBackOffPolicy backOffPolicy = new NoBackOffPolicy();
@@ -372,4 +375,20 @@ JdbcTransactionManager transactionManager(DataSource dataSource) {
 
 	}
 
+	public static class AnnotationBasedChunkListener {
+
+		@BeforeChunk
+		public void beforeChunk() {
+		}
+
+		@AfterChunk
+		public void afterChunk() {
+		}
+
+		@AfterChunkError
+		public void afterChunkError() {
+		}
+
+	}
+
 }