Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-2445: make annotation based chunk listeners available to non-fault tolerant steps

Browse Source 
Before this commit, annotation based chunk listeners were not registered
when using a non-fault tolerant step builder.

This commit moves the code of chunk listener annotations handling to the
AbstractTaskletStepBuilder so that other tasklet builders can use it.

Resolves BATCH-2445
This commit is contained in:
Mahmoud Ben Hassine
2018-02-01 18:27:47 +01:00
committed by Michael Minella
parent e11b838452
commit 1ed5a33bd1
3 changed files with 149 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 the original author or authors.
* Copyright 2012-2018 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,12 +15,18 @@
*/
package org.springframework.batch.core.step.builder;
import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Set;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
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.StepListenerFactoryBean;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemStream;
@@ -29,6 +35,7 @@ import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.batch.support.ReflectionUtils;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.interceptor.TransactionAttribute;
@@ -39,6 +46,7 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
*
* @author Dave Syer
* @author Michael Minella
* @author Mahmoud Ben Hassine
*
* @since 2.2
*
@@ -136,6 +144,32 @@ StepBuilderHelper<AbstractTaskletStepBuilder<B>> {
return this;
}
/**
* Registers objects using the annotation based listener configuration.
*
* @param listener the object that has a method configured with listener annotation
* @return this for fluent chaining
*/
@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.size() > 0) {
StepListenerFactoryBean factory = new StepListenerFactoryBean();
factory.setDelegate(listener);
this.listener((ChunkListener) factory.getObject());
}
@SuppressWarnings("unchecked")
B result = (B) this;
return result;
}
/**
* Register a stream for callbacks that manage restart data.
*

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

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2014 the original author or authors.
* Copyright 2006-2018 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.
@@ -32,9 +32,6 @@ import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
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.annotation.OnSkipInProcess;
import org.springframework.batch.core.annotation.OnSkipInRead;
import org.springframework.batch.core.annotation.OnSkipInWrite;
@@ -93,6 +90,7 @@ import org.springframework.util.Assert;
* @author Dave Syer
* @author Chris Schaefer
* @author Michael Minella
* @author Mahmoud Ben Hassine
*
* @since 2.2
*/
@@ -206,23 +204,12 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
skipListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnSkipInProcess.class));
skipListenerMethods.addAll(ReflectionUtils.findMethod(listener.getClass(), OnSkipInWrite.class));
Set<Method> chunkListenerMethods = new HashSet<Method>();
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(skipListenerMethods.size() > 0) {
StepListenerFactoryBean factory = new StepListenerFactoryBean();
factory.setDelegate(listener);
skipListeners.add((SkipListener) factory.getObject());
}
if(chunkListenerMethods.size() > 0) {
StepListenerFactoryBean factory = new StepListenerFactoryBean();
factory.setDelegate(listener);
super.listener(new TerminateOnExceptionChunkListenerDelegate((ChunkListener) factory.getObject()));
}
@SuppressWarnings("unchecked")
SimpleStepBuilder<I, O> result = this;
return result;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2014 the original author or authors.
* Copyright 2012-2018 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.
@@ -27,6 +27,7 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.annotation.AfterChunk;
import org.springframework.batch.core.annotation.AfterChunkError;
import org.springframework.batch.core.annotation.AfterProcess;
import org.springframework.batch.core.annotation.AfterRead;
import org.springframework.batch.core.annotation.AfterStep;
@@ -36,7 +37,9 @@ 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.configuration.xml.DummyItemReader;
import org.springframework.batch.core.configuration.xml.DummyItemWriter;
import org.springframework.batch.core.job.SimpleJob;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.item.ItemReader;
@@ -51,6 +54,7 @@ import static org.junit.Assert.assertEquals;
/**
* @author Dave Syer
* @author Michael Minella
* @author Mahmoud Ben Hassine
*
*/
@SuppressWarnings("serial")
@@ -87,6 +91,85 @@ public class StepBuilderTests {
assertEquals(1, InterfaceBasedStepExecutionListener.afterStepCount);
assertEquals(1, AnnotationBasedStepExecutionListener.beforeStepCount);
assertEquals(1, AnnotationBasedStepExecutionListener.afterStepCount);
assertEquals(1, AnnotationBasedStepExecutionListener.beforeChunkCount);
assertEquals(1, AnnotationBasedStepExecutionListener.afterChunkCount);
}
@Test
public void testAnnotationBasedChunkListenerForTaskletStep() throws Exception {
JobRepository jobRepository = new MapJobRepositoryFactoryBean().getObject();
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());
builder.build().execute(execution);
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
assertEquals(1, AnnotationBasedChunkListener.beforeChunkCount);
assertEquals(1, AnnotationBasedChunkListener.afterChunkCount);
}
@Test
public void testAnnotationBasedChunkListenerForSimpleTaskletStep() throws Exception {
JobRepository jobRepository = new MapJobRepositoryFactoryBean().getObject();
StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters()).createStepExecution("step");
jobRepository.add(execution);
PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
SimpleStepBuilder builder = new StepBuilder("step")
.repository(jobRepository)
.transactionManager(transactionManager)
.chunk(5)
.reader(new DummyItemReader())
.writer(new DummyItemWriter())
.listener(new AnnotationBasedChunkListener());
builder.build().execute(execution);
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
assertEquals(1, AnnotationBasedChunkListener.beforeChunkCount);
assertEquals(1, AnnotationBasedChunkListener.afterChunkCount);
}
@Test
public void testAnnotationBasedChunkListenerForFaultTolerantTaskletStep() throws Exception {
JobRepository jobRepository = new MapJobRepositoryFactoryBean().getObject();
StepExecution execution = jobRepository.createJobExecution("foo", new JobParameters()).createStepExecution("step");
jobRepository.add(execution);
PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
SimpleStepBuilder builder = new StepBuilder("step")
.repository(jobRepository)
.transactionManager(transactionManager)
.chunk(5)
.reader(new DummyItemReader())
.writer(new DummyItemWriter())
.faultTolerant()
.listener(new AnnotationBasedChunkListener()); // TODO should this return FaultTolerantStepBuilder?
builder.build().execute(execution);
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
assertEquals(1, AnnotationBasedChunkListener.beforeChunkCount);
assertEquals(1, AnnotationBasedChunkListener.afterChunkCount);
}
@Test
public void testAnnotationBasedChunkListenerForJobStepBuilder() throws Exception {
JobRepository jobRepository = new MapJobRepositoryFactoryBean().getObject();
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)
.transactionManager(transactionManager)
.job(job)
.listener(new AnnotationBasedChunkListener());
builder.build().execute(execution);
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
// it makes no sense to register a ChunkListener on a step which is not of type tasklet, so it should not be invoked
assertEquals(0, AnnotationBasedChunkListener.beforeChunkCount);
assertEquals(0, AnnotationBasedChunkListener.afterChunkCount);
}
@Test
@@ -258,4 +341,32 @@ public class StepBuilderTests {
afterChunkCount++;
}
}
public static class AnnotationBasedChunkListener {
static int beforeChunkCount = 0;
static int afterChunkCount = 0;
static int afterChunkErrorCount = 0;
public AnnotationBasedChunkListener() {
beforeChunkCount = 0;
afterChunkCount = 0;
afterChunkErrorCount = 0;
}
@BeforeChunk
public void beforeChunk() {
beforeChunkCount++;
}
@AfterChunk
public void afterChunk() {
afterChunkCount++;
}
@AfterChunkError
public void afterChunkError() {
afterChunkErrorCount++;
}
}
}