diff --git a/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java index 20d7f12a6..0c544291a 100644 --- a/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java +++ b/spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java @@ -18,6 +18,7 @@ package org.springframework.batch.repeat.support; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.Collections; @@ -35,6 +36,7 @@ import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** @@ -48,7 +50,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; */ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { - static Log logger = LogFactory.getLog(TaskExecutorRepeatTemplateBulkAsynchronousTests.class); + static Log logger = LogFactory + .getLog(TaskExecutorRepeatTemplateBulkAsynchronousTests.class); private int total = 1000; @@ -56,23 +59,26 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { private volatile int early = Integer.MAX_VALUE; + private volatile int error = Integer.MAX_VALUE; + private TaskExecutorRepeatTemplate template; private RepeatCallback callback; private List items; - private ThreadPoolTaskExecutor taskExecutor; + private ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor(); @Before public void setUp() { template = new TaskExecutorRepeatTemplate(); - taskExecutor = new ThreadPoolTaskExecutor(); - taskExecutor.setMaxPoolSize(300); - taskExecutor.setCorePoolSize(10); - taskExecutor.setQueueCapacity(0); - taskExecutor.afterPropertiesSet(); + TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + threadPool.setMaxPoolSize(300); + threadPool.setCorePoolSize(10); + threadPool.setQueueCapacity(0); + threadPool.afterPropertiesSet(); + taskExecutor = threadPool; template.setTaskExecutor(taskExecutor); template.setThrottleLimit(throttleLimit); @@ -82,7 +88,8 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { private volatile AtomicInteger count = new AtomicInteger(0); - public RepeatStatus doInIteration(RepeatContext context) throws Exception { + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { int position = count.incrementAndGet(); String item = position <= total ? "" + position : null; items.add("" + item); @@ -96,9 +103,14 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { * happens for instance if there is a failure and you want to * retry the work.) */ - RepeatStatus result = RepeatStatus.continueIf(position != early && item != null); + RepeatStatus result = RepeatStatus.continueIf(position != early + && item != null); + if (position == error) { + throw new RuntimeException("Planned"); + } if (!result.isContinuable()) { - logger.debug("Returning " + result + " for count=" + position); + logger.debug("Returning " + result + " for count=" + + position); } return result; } @@ -106,6 +118,11 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { } + @After + public void tearDown() { + threadPool.destroy(); + } + @Test public void testThrottleLimit() throws Exception { @@ -119,11 +136,6 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { } - @After - public void tearDown() { - taskExecutor.destroy(); - } - @Test public void testThrottleLimitEarlyFinish() throws Exception { @@ -151,7 +163,8 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { taskExecutor.setQueueCapacity(0); // This is the most sensible setting, otherwise the bookkeeping in // ResultHolderResultQueue gets out of whack when tasks are aborted. - taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + taskExecutor + .setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); taskExecutor.afterPropertiesSet(); template.setTaskExecutor(taskExecutor); @@ -162,7 +175,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { // Extra tasks will be submitted before the termination is detected assertEquals(total, items.size() - frequency); assertTrue(frequency <= throttleLimit + 1); - + taskExecutor.destroy(); } @@ -204,6 +217,22 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { } + @Test + public void testThrottleLimitWithError() throws Exception { + + error = 50; + + try { + template.iterate(callback); + fail("Expected planned exception"); + } catch (Exception e) { + assertEquals("Planned", e.getMessage()); + } + int frequency = Collections.frequency(items, "null"); + assertEquals(0, frequency); + + } + /** * Slightly flakey convenience method. If this doesn't do something that * lasts sufficiently long for another worker to be launched while it is 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 4825d10d3..9c50fe589 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 @@ -167,7 +167,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { */ ResultHolder future; try { - future = (ResultHolder) queue.take(); + future = queue.take(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java deleted file mode 100644 index dd48719e6..000000000 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java +++ /dev/null @@ -1,201 +0,0 @@ -/* - * Copyright 2006-2007 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.repeat.support; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.Before; -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.policy.SimpleCompletionPolicy; -import org.springframework.core.task.SimpleAsyncTaskExecutor; - -/** - * Simple tests for concurrent behaviour in repeat template, in particular the - * barrier at the end of the iteration. N.B. these tests may fail if - * insufficient threads are available (e.g. on a single-core machine, or under - * load). They shouldn't deadlock though. - * - * @author Dave Syer - * - */ -public class TaskExecutorRepeatTemplateBulkAsynchronousTests { - - static Log logger = LogFactory.getLog(TaskExecutorRepeatTemplateBulkAsynchronousTests.class); - - private int total = 20; - - private int throttleLimit = 8; - - private volatile int early = Integer.MAX_VALUE; - - private TaskExecutorRepeatTemplate template; - - private RepeatCallback callback; - - private List items; - - @Before - public void setUp() { - - template = new TaskExecutorRepeatTemplate(); - SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(300); - template.setTaskExecutor(taskExecutor); - template.setThrottleLimit(throttleLimit); - - items = Collections.synchronizedList(new ArrayList()); - - callback = new RepeatCallback() { - - private volatile AtomicInteger count = new AtomicInteger(0); - - public RepeatStatus doInIteration(RepeatContext context) throws Exception { - int position = count.incrementAndGet(); - String item = position <= total ? "" + position : null; - items.add("" + item); - if (item != null) { - beBusy(); - } - /* - * In a multi-threaded task, one of the callbacks can call - * FINISHED early, while other threads are still working, and - * would do more work if the callback was called again. (This - * happens for instance if there is a failure and you want to - * retry the work.) - */ - RepeatStatus result = RepeatStatus.continueIf(position != early && item != null); - if (!result.isContinuable()) { - logger.debug("Returning " + result + " for count=" + position); - } - return result; - } - }; - - } - - @Test - public void testThrottleLimit() throws Exception { - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - // System.err.println(items); - // System.err.println("Frequency: " + frequency); - assertEquals(total, items.size() - frequency); - assertTrue(frequency > 1); - assertTrue(frequency <= throttleLimit + 1); - - } - - @Test - public void testThrottleLimitEarlyFinish() throws Exception { - - early = 2; - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - // System.err.println("Frequency: " + frequency); - // System.err.println("Items: " + items); - assertEquals(total, items.size() - frequency); - assertTrue(frequency > 1); - assertTrue(frequency <= throttleLimit + 1); - - } - - @Test - public void testThrottleLimitEarlyFinishThreadStarvation() throws Exception { - - early = 2; - SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - // Set the concurrency limit below the throttle limit for possible - // starvation condition - taskExecutor.setConcurrencyLimit(20); - template.setTaskExecutor(taskExecutor); - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - // System.err.println("Frequency: " + frequency); - // System.err.println("Items: " + items); - // Extra tasks will be submitted before the termination is detected - assertEquals(total, items.size() - frequency); - assertTrue(frequency <= throttleLimit + 1); - - } - - @Test - public void testThrottleLimitEarlyFinishOneThread() throws Exception { - - early = 4; - SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(1); - - // This is kind of slow with only one thread, so reduce size: - throttleLimit = 4; - total = 10; - - template.setThrottleLimit(throttleLimit); - template.setTaskExecutor(taskExecutor); - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - // System.err.println("Frequency: " + frequency); - // System.err.println("Items: " + items); - assertEquals(total, items.size() - frequency); - assertTrue(frequency <= throttleLimit + 1); - - } - - @Test - public void testThrottleLimitWithEarlyCompletion() throws Exception { - - early = 2; - template.setCompletionPolicy(new SimpleCompletionPolicy(10)); - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - assertEquals(10, items.size() - frequency); - // System.err.println("Frequency: " + frequency); - assertEquals(0, frequency); - - } - - /** - * Slightly flakey convenience method. If this doesn't do something that - * lasts sufficiently long for another worker to be launched while it is - * busy, the early completion tests will fail. "Sufficiently long" is the - * problem so we try and block until we know someone else is busy? - * - * @throws Exception - */ - private void beBusy() throws Exception { - synchronized (this) { - wait(100L); - notifyAll(); - } - } - -}