diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/RepeatTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/RepeatTemplate.java index 47855b360..da1311321 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/RepeatTemplate.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/RepeatTemplate.java @@ -183,7 +183,10 @@ public class RepeatTemplate implements RepeatOperations { RepeatStatus result = RepeatStatus.CONTINUABLE; RepeatInternalState state = createInternalState(context); + // This is the list of exceptions thrown by all active callbacks Collection throwables = state.getThrowables(); + // Keep a separate list of exceptions we handled that need to be rethrown + Collection deferred = new ArrayList(); try { @@ -214,33 +217,11 @@ public class RepeatTemplate implements RepeatOperations { } catch (Throwable throwable) { - - // An exception alone is not sufficient grounds for not - // continuing - Throwable unwrappedThrowable = unwrapIfRethrown(throwable); - try { - - for (int i = listeners.length; i-- > 0;) { - RepeatListener interceptor = listeners[i]; - // This is not an error - only log at debug - // level. - logger.debug("Exception intercepted (" + (i + 1) + " of " + listeners.length + ")", - unwrappedThrowable); - interceptor.onError(context, unwrappedThrowable); - } - - logger.debug("Handling exception: " + throwable.getClass().getName() + ", caused by: " - + unwrappedThrowable.getClass().getName() + ": " + unwrappedThrowable.getMessage()); - exceptionHandler.handleException(context, unwrappedThrowable); - - } - catch (Throwable handled) { - throwables.add(handled); - } + doHandle(throwable, context, deferred); } // N.B. the order may be important here: - if (isComplete(context, result) || isMarkedComplete(context) || !throwables.isEmpty()) { + if (isComplete(context, result) || isMarkedComplete(context) || !deferred.isEmpty()) { running = false; } } @@ -248,6 +229,9 @@ public class RepeatTemplate implements RepeatOperations { } result = result.and(waitForResults(state)); + for (Throwable throwable : throwables) { + doHandle(throwable, context, deferred); + } // Explicitly drop any references to internal state... state = null; @@ -262,9 +246,9 @@ public class RepeatTemplate implements RepeatOperations { try { - if (!throwables.isEmpty()) { - Throwable throwable = (Throwable) throwables.iterator().next(); - logger.debug("Handling fatal exception explicitly (rethrowing first of " + throwables.size() + if (!deferred.isEmpty()) { + Throwable throwable = (Throwable) deferred.iterator().next(); + logger.debug("Handling fatal exception explicitly (rethrowing first of " + deferred.size() + "): " + throwable.getClass().getName() + ": " + throwable.getMessage()); rethrow(throwable); } @@ -291,6 +275,32 @@ public class RepeatTemplate implements RepeatOperations { } + private void doHandle(Throwable throwable, RepeatContext context, + Collection deferred) { + // An exception alone is not sufficient grounds for not + // continuing + Throwable unwrappedThrowable = unwrapIfRethrown(throwable); + try { + + for (int i = listeners.length; i-- > 0;) { + RepeatListener interceptor = listeners[i]; + // This is not an error - only log at debug + // level. + logger.debug("Exception intercepted (" + (i + 1) + " of " + listeners.length + ")", + unwrappedThrowable); + interceptor.onError(context, unwrappedThrowable); + } + + logger.debug("Handling exception: " + throwable.getClass().getName() + ", caused by: " + + unwrappedThrowable.getClass().getName() + ": " + unwrappedThrowable.getMessage()); + exceptionHandler.handleException(context, unwrappedThrowable); + + } + catch (Throwable handled) { + deferred.add(handled); + } + } + /** * Re-throws the original throwable if it is unchecked, wraps checked * exceptions into {@link RepeatException}. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java index 69537247a..49a25313c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java @@ -120,7 +120,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { /* * N.B. If the queue is empty then take() blocks until a result appears, - * and there must be at least one because we just submitted one to teh + * and there must be at least one because we just submitted one to the * task executor. */ ResultHolder result = queue.take(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java index df400f8c8..5c77387eb 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcCursorItemReaderConfigTests.java @@ -1,6 +1,10 @@ package org.springframework.batch.item.database; -import static org.easymock.EasyMock.*; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.createNiceMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; import java.sql.Connection; import java.sql.PreparedStatement; @@ -29,14 +33,13 @@ public class JdbcCursorItemReaderConfigTests { DataSource ds = createMock(DataSource.class); Connection con = createMock(Connection.class); expect(con.getAutoCommit()).andReturn(false); - PreparedStatement ps = createMock(PreparedStatement.class); + PreparedStatement ps = createNiceMock(PreparedStatement.class); expect(con.prepareStatement("select foo from bar", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, ResultSet.HOLD_CURSORS_OVER_COMMIT)).andReturn(ps); expect(ds.getConnection()).andReturn(con); expect(ds.getConnection()).andReturn(con); con.commit(); - replay(con); - replay(ds); + replay(con, ds, ps); PlatformTransactionManager tm = new DataSourceTransactionManager(ds); TransactionTemplate tt = new TransactionTemplate(tm); final JdbcCursorItemReader reader = new JdbcCursorItemReader(); @@ -64,13 +67,12 @@ public class JdbcCursorItemReaderConfigTests { DataSource ds = createMock(DataSource.class); Connection con = createMock(Connection.class); expect(con.getAutoCommit()).andReturn(false); - PreparedStatement ps = createMock(PreparedStatement.class); + PreparedStatement ps = createNiceMock(PreparedStatement.class); expect(con.prepareStatement("select foo from bar", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)).andReturn(ps); expect(ds.getConnection()).andReturn(con); expect(ds.getConnection()).andReturn(con); con.commit(); - replay(con); - replay(ds); + replay(con, ds, ps); PlatformTransactionManager tm = new DataSourceTransactionManager(ds); TransactionTemplate tt = new TransactionTemplate(tm); final JdbcCursorItemReader reader = new JdbcCursorItemReader(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/SimpleRepeatTemplateTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/SimpleRepeatTemplateTests.java index 54f28e55d..2f4797b7e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/SimpleRepeatTemplateTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/SimpleRepeatTemplateTests.java @@ -50,7 +50,10 @@ public class SimpleRepeatTemplateTests extends AbstractTradeBatchTests { int count = 0; public RepeatTemplate getRepeatTemplate() { - return new RepeatTemplate(); + template = new RepeatTemplate(); + // default stop after more items than exist in dataset + template.setCompletionPolicy(new SimpleCompletionPolicy(8)); + return template; } @Test @@ -97,6 +100,7 @@ public class SimpleRepeatTemplateTests extends AbstractTradeBatchTests { } assertEquals(1, count); + assertTrue("Too many attempts: "+count, count<=10); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AsynchronousRepeatTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java similarity index 64% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AsynchronousRepeatTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java index a32ea96bb..59b0ac304 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AsynchronousRepeatTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java @@ -19,6 +19,7 @@ package org.springframework.batch.repeat.support; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.Collections; import java.util.HashSet; @@ -28,9 +29,75 @@ import org.junit.Test; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.batch.repeat.exception.ExceptionHandler; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.core.task.SimpleAsyncTaskExecutor; -public class AsynchronousRepeatTests extends AbstractTradeBatchTests { +public class TaskExecutorRepeatTemplateAsynchronousTests extends + AbstractTradeBatchTests { + + RepeatTemplate template = getRepeatTemplate(); + + int count = 0; + + // @Override + public RepeatTemplate getRepeatTemplate() { + TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + template.setTaskExecutor(new SimpleAsyncTaskExecutor()); + return template; + } + + @Test + public void testEarlyCompletionWithException() throws Exception { + + TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + template.setCompletionPolicy(new SimpleCompletionPolicy(20)); + taskExecutor.setConcurrencyLimit(2); + template.setTaskExecutor(taskExecutor); + try { + template.iterate(new RepeatCallback() { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { + count++; + throw new IllegalStateException("foo!"); + } + }); + fail("Expected IllegalStateException"); + } catch (IllegalStateException e) { + assertEquals("foo!", e.getMessage()); + } + + assertTrue("Too few attempts: "+count, count>=2); + assertTrue("Too many attempts: "+count, count<=10); + + } + + @Test + public void testExceptionHandlerSwallowsException() throws Exception { + + TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + template.setCompletionPolicy(new SimpleCompletionPolicy(4)); + taskExecutor.setConcurrencyLimit(2); + template.setTaskExecutor(taskExecutor); + + template.setExceptionHandler(new ExceptionHandler() { + public void handleException(RepeatContext context, + Throwable throwable) throws Throwable { + count++; + } + }); + template.iterate(new RepeatCallback() { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { + throw new IllegalStateException("foo!"); + } + }); + + assertEquals(4, count); + + } /** * Run a batch with a single template that itself has an async task @@ -42,22 +109,20 @@ public class AsynchronousRepeatTests extends AbstractTradeBatchTests { @Test public void testMultiThreadAsynchronousExecution() throws Exception { - TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); - template.setTaskExecutor(new SimpleAsyncTaskExecutor()); - final String threadName = Thread.currentThread().getName(); final Set threadNames = new HashSet(); final RepeatCallback callback = new RepeatCallback() { - public RepeatStatus doInIteration(RepeatContext context) throws Exception { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { assertNotSame(threadName, Thread.currentThread().getName()); threadNames.add(Thread.currentThread().getName()); Thread.sleep(100); Trade item = provider.read(); - if (item!=null) { + if (item != null) { processor.write(Collections.singletonList(item)); } - return RepeatStatus.continueIf(item!=null); + return RepeatStatus.continueIf(item != null); } }; @@ -67,7 +132,7 @@ public class AsynchronousRepeatTests extends AbstractTradeBatchTests { assertEquals(NUMBER_OF_ITEMS, processor.count); assertTrue(threadNames.size() > 1); } - + @Test public void testThrottleLimit() throws Exception { TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); @@ -80,14 +145,15 @@ public class AsynchronousRepeatTests extends AbstractTradeBatchTests { final Set threadNames = new HashSet(); final RepeatCallback callback = new RepeatCallback() { - public RepeatStatus doInIteration(RepeatContext context) throws Exception { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { assertNotSame(threadName, Thread.currentThread().getName()); threadNames.add(Thread.currentThread().getName()); Trade item = provider.read(); - if (item!=null) { + if (item != null) { processor.write(Collections.singletonList(item)); } - return RepeatStatus.continueIf(item!=null); + return RepeatStatus.continueIf(item != null); } }; @@ -115,8 +181,10 @@ public class AsynchronousRepeatTests extends AbstractTradeBatchTests { final String threadName = Thread.currentThread().getName(); final Set threadNames = new HashSet(); - final RepeatCallback stepCallback = new ItemReaderRepeatCallback(provider, processor) { - public RepeatStatus doInIteration(RepeatContext context) throws Exception { + final RepeatCallback stepCallback = new ItemReaderRepeatCallback( + provider, processor) { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { assertNotSame(threadName, Thread.currentThread().getName()); threadNames.add(Thread.currentThread().getName()); Thread.sleep(100); @@ -124,7 +192,8 @@ public class AsynchronousRepeatTests extends AbstractTradeBatchTests { } }; RepeatCallback jobCallback = new RepeatCallback() { - public RepeatStatus doInIteration(RepeatContext context) throws Exception { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { stepTemplate.iterate(stepCallback); return RepeatStatus.FINISHED; }