diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java index 90246b0cd..8107378c8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java @@ -176,7 +176,7 @@ public class RetryTemplate implements RetryOperations { retryPolicy.registerThrowable(context, throwable); if (retryPolicy.shouldRethrow(context)) { - logger.debug("Abort retry for policy: count=" + context.getRetryCount()); + logger.debug("Rethrow in retry for policy: count=" + context.getRetryCount()); rethrow(throwable); } diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml index d610cdabb..916f93204 100644 --- a/spring-batch-integration/pom.xml +++ b/spring-batch-integration/pom.xml @@ -1,218 +1,206 @@ - - - 4.0.0 - spring-batch-integration - Enterprise Integration - jar - - org.springframework.batch - spring-batch - 1.1.0.DEV-SNAPSHOT - .. - - - 1.0.0.CI-SNAPSHOT - - - - default - - true - - - - - - org.apache.maven.plugins - maven-surefire-plugin - - true - - - - - - - test - - - - - org.apache.maven.plugins - maven-surefire-plugin - - false - - - - - - - - - javax.jms - com.springsource.javax.jms - 1.1.0 - compile - - - junit - junit - 4.4 - test - - - cglib - cglib-nodep - test - - - org.springframework.integration - org.springframework.integration - ${spring.integration.version} - compile - - - org.springframework - org.springframework.context - - - org.springframework - org.springframework.aop - - - org.springframework - org.springframework.transaction - - - - - org.springframework.integration - org.springframework.integration.adapter - ${spring.integration.version} - compile - - - org.springframework - org.springframework.web - - - org.springframework - org.springframework.aop - - - org.springframework - org.springframework.context - - - org.springframework - org.springframework.context.support - - - org.springframework - org.springframework.jms - - - org.springframework - org.springframework.transaction - - - org.apache.commons - com.springsource.org.apache.commons.net - - - org.apache.oro - com.springsource.org.apache.oro - - - - - org.springframework - spring-context - ${spring.framework.version} - compile - - - org.springframework - spring-aop - ${spring.framework.version} - compile - - - org.springframework - spring-jms - ${spring.framework.version} - compile - - - org.springframework - spring-tx - ${spring.framework.version} - compile - - - org.springframework - spring-test - ${spring.framework.version} - test - - - org.springframework.batch - spring-batch-core - ${project.version} - compile - - - org.springframework - spring-context - - - org.springframework - spring-beans - - - org.springframework - spring-core - - - org.springframework - spring-aop - - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.5 - 1.5 - - - - - - - springsource-external - http://repository.springsource.com/maven/bundles/external - - - springsource-release - http://repository.springsource.com/maven/bundles/release - - - springsource-milestone - http://repository.springsource.com/maven/bundles/milestone - - - - springsource-snapshot - http://repository.springsource.com/maven/bundles/snapshot - - - + + + spring-batch + org.springframework.batch + 1.1.0.DEV-SNAPSHOT + .. + + 4.0.0 + spring-batch-integration + Enterprise Integration + + + + maven-compiler-plugin + + 1.5 + 1.5 + + + + + + + + true + + + + + maven-surefire-plugin + + true + + + + + + + test + + + + maven-surefire-plugin + + false + + + + + + + + + springsource-external + http://repository.springsource.com/maven/bundles/external + + + springsource-release + http://repository.springsource.com/maven/bundles/release + + + springsource-milestone + http://repository.springsource.com/maven/bundles/milestone + + + springsource-snapshot + http://repository.springsource.com/maven/bundles/snapshot + + + + + javax.jms + com.springsource.javax.jms + 1.1.0 + compile + + + junit + junit + 4.4 + test + + + cglib + cglib-nodep + test + + + org.springframework.integration + org.springframework.integration + ${spring.integration.version} + compile + + + org.springframework.context + org.springframework + + + org.springframework.aop + org.springframework + + + org.springframework.transaction + org.springframework + + + + + org.springframework.integration + org.springframework.integration.adapter + ${spring.integration.version} + compile + + + org.springframework.web + org.springframework + + + org.springframework.aop + org.springframework + + + org.springframework.context + org.springframework + + + org.springframework.context.support + org.springframework + + + org.springframework.jms + org.springframework + + + org.springframework.transaction + org.springframework + + + com.springsource.org.apache.commons.net + org.apache.commons + + + com.springsource.org.apache.oro + org.apache.oro + + + + + org.springframework + spring-context + ${spring.framework.version} + compile + + + org.springframework + spring-aop + ${spring.framework.version} + compile + + + org.springframework + spring-jms + ${spring.framework.version} + compile + + + org.springframework + spring-tx + ${spring.framework.version} + compile + + + org.springframework + spring-test + ${spring.framework.version} + test + + + org.springframework.batch + spring-batch-core + ${project.version} + compile + + + spring-context + org.springframework + + + spring-beans + org.springframework + + + spring-core + org.springframework + + + spring-aop + org.springframework + + + + + + 1.0.0.CI-SNAPSHOT + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java index d80c97fea..0dafec109 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java @@ -19,6 +19,9 @@ import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; @@ -31,6 +34,7 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.dispatcher.DirectChannel; import org.springframework.integration.endpoint.EndpointTrigger; import org.springframework.integration.endpoint.SourceEndpoint; +import org.springframework.integration.endpoint.interceptor.EndpointInterceptorAdapter; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSource; @@ -38,7 +42,7 @@ import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.SimpleTaskScheduler; import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.integration.util.ErrorHandler; +import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.StringUtils; @@ -68,6 +72,8 @@ public class PollableSourceRetryTests { } }; + private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();; + @SuppressWarnings("unchecked") @Test public void testSimpleTransactionalPolling() throws Exception { @@ -86,10 +92,10 @@ public class PollableSourceRetryTests { MessageSource source = getPollableSource(list); DirectChannel channel = getChannel(handler, source); SourceEndpoint endpoint = getSourceEndpoint(source, channel); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll"); + addTransactionInterceptor(endpoint); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 2, 40); @@ -155,10 +161,10 @@ public class PollableSourceRetryTests { MessageSource source = getPollableSource(list); DirectChannel channel = getChannel(handler, source); SourceEndpoint endpoint = getSourceEndpoint(source, channel); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll"); + addTransactionInterceptor(endpoint); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 2, 20); @@ -197,13 +203,13 @@ public class PollableSourceRetryTests { MessageSource source = getPollableSource(list); DirectChannel channel = getChannel(handler, source); SourceEndpoint endpoint = getSourceEndpoint(source, channel); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll"); + addTransactionInterceptor(endpoint); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); - waitForResults(scheduler, 5, 20); + waitForResults(scheduler, 5, 30); assertEquals(5, processed.size()); assertFalse("No messages got to processor", processed.isEmpty()); @@ -243,10 +249,11 @@ public class PollableSourceRetryTests { DirectChannel channel = getChannel(handler, source); SourceEndpoint endpoint = getSourceEndpoint(source, channel); // endpoint.addInterceptor(getTransactionInterceptor()); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getRepeatOperationsInterceptor(3), getTransactionInterceptor()}, "poll"); + addTransactionInterceptor(endpoint); + addRepeatInterceptor(endpoint, 3); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 6, 100); @@ -292,17 +299,17 @@ public class PollableSourceRetryTests { MessageSource source = getPollableSource(list); MessageChannel channel = getChannel(handler, source); - SourceEndpoint endpoint = getSourceEndpoint(source, channel); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll"); // this was the old dispatch advice chain channel = (MessageChannel) getProxy(channel, MessageChannel.class, new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }, "send"); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + addTransactionInterceptor(endpoint); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); - waitForResults(scheduler, 4, 20); + waitForResults(scheduler, 4, 40); assertEquals(4, processed.size()); assertEquals(1, recovered.size()); @@ -340,14 +347,15 @@ public class PollableSourceRetryTests { MessageSource source = getPollableSource(list); MessageChannel channel = getChannel(handler, source); - SourceEndpoint endpoint = getSourceEndpoint(source, channel); - MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getRepeatOperationsInterceptor(3), getTransactionInterceptor()}, "poll"); // this was the old dispatch advice chain channel = (MessageChannel) getProxy(channel, MessageChannel.class, new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }, "send"); + SourceEndpoint endpoint = getSourceEndpoint(source, channel); + addTransactionInterceptor(endpoint); + addRepeatInterceptor(endpoint, 3); endpoint.afterPropertiesSet(); EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule()); - trigger.addTarget(target); + trigger.addTarget(endpoint); TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger); waitForResults(scheduler, 6, 100); @@ -373,7 +381,7 @@ public class PollableSourceRetryTests { * @return */ private SourceEndpoint getSourceEndpoint(MessageSource source, MessageChannel channel) { - PollingSchedule schedule = new PollingSchedule(50); + PollingSchedule schedule = new PollingSchedule(100); schedule.setFixedRate(true); // used to be the default SourceEndpoint endpoint = new SourceEndpoint(source); endpoint.setOutputChannel(channel); @@ -426,15 +434,15 @@ public class PollableSourceRetryTests { return source; } - // Workaround for INT-182 private TaskScheduler getSchedulerWithErrorHandler(Runnable task) { SimpleTaskScheduler scheduler = new SimpleTaskScheduler(Executors.newSingleThreadScheduledExecutor()); - scheduler.setErrorHandler(new ErrorHandler() { - public void handle(Throwable t) { - logger.error("Exception in scheduler", t); - // throw (RuntimeException)t; - } - }); + // Workaround for INT-182 +// scheduler.setErrorHandler(new ErrorHandler() { +// public void handle(Throwable t) { +// logger.error("Exception in scheduler", t); +// // throw (RuntimeException)t; +// } +// }); scheduler.schedule(task); return scheduler; } @@ -454,7 +462,9 @@ public class PollableSourceRetryTests { if (data.getClass().isArray()) { data = ((Object[]) data)[0]; } - recovered.add(((Message) data).getPayload()); + String payload = ((Message) data).getPayload(); + logger.debug("Recovering: "+payload); + recovered.add(payload); return true; } }); @@ -462,11 +472,45 @@ public class PollableSourceRetryTests { return advice; } + /** + * @param endpoint + */ + private void addTransactionInterceptor(SourceEndpoint endpoint) { + org.springframework.integration.endpoint.interceptor.TransactionInterceptor transactionInterceptor = new org.springframework.integration.endpoint.interceptor.TransactionInterceptor(transactionManager); + transactionInterceptor.afterPropertiesSet(); + endpoint.addInterceptor(transactionInterceptor); + } + /** * @return */ private TransactionInterceptor getTransactionInterceptor() { - return new TransactionInterceptor(new ResourcelessTransactionManager(), PropertiesConverter.stringToProperties("*=PROPAGATION_REQUIRED")); + return new TransactionInterceptor(transactionManager, PropertiesConverter.stringToProperties("*=PROPAGATION_REQUIRED")); + } + + /** + * @param endpoint + * @param commitInterval + */ + private void addRepeatInterceptor(SourceEndpoint endpoint, int commitInterval) { + final RepeatTemplate repeatTemplate = new RepeatTemplate(); + repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); + endpoint.addInterceptor(new EndpointInterceptorAdapter() { + private boolean value; + @Override + public boolean aroundSend(final Message message, final MessageTarget endpoint) { + repeatTemplate.iterate(new RepeatCallback() { + public ExitStatus doInIteration(RepeatContext context) throws Exception { + doAroundSend(message, endpoint); + return ExitStatus.CONTINUABLE; + } + }); + return value; + } + private void doAroundSend(Message message, MessageTarget endpoint) { + value = super.aroundSend(message, endpoint); + } + }); } /**