diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java index 0179eb969..9b3bb17f1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-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. @@ -56,10 +56,13 @@ import org.springframework.util.Assert; * cumulative result. * * @author Michael Minella + * @author Mahmoud Ben Hassine * @since 3.0 */ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { + private static final int DEFAULT_POLLING_INTERVAL = 500; + // TODO: Replace with proper Channel and Messages once minimum support level for Spring is 4 private Queue partitionDataQueue; private ReentrantLock lock; @@ -72,6 +75,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { private JobRepository jobRepository; private boolean allowStartIfComplete = false; private Set partitionStepNames = new HashSet(); + private int pollingInterval = DEFAULT_POLLING_INTERVAL; /** * @return the step that will be executed by each partition @@ -156,6 +160,14 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { this.jobRepository = jobRepository; } + /** + * @param pollingInterval the duration of partitions completion polling interval + * (in milliseconds). The default value is 500ms. + */ + public void setPollingInterval(int pollingInterval) { + this.pollingInterval = pollingInterval; + } + /* (non-Javadoc) * @see org.springframework.batch.core.partition.PartitionHandler#handle(org.springframework.batch.core.partition.StepExecutionSplitter, org.springframework.batch.core.StepExecution) */ @@ -224,6 +236,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { final List> tasks, final Set result) throws Exception { while(true) { + Thread.sleep(pollingInterval); try { lock.lock(); while(!partitionDataQueue.isEmpty()) { @@ -385,6 +398,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { Assert.notNull(propertyContext, "A BatchPropertyContext is required"); Assert.isTrue(mapper != null || (threads > 0 || partitions > 0), "Either a mapper implementation or the number of partitions/threads is required"); Assert.notNull(jobRepository, "A JobRepository is required"); + Assert.isTrue(pollingInterval >= 0, "The polling interval must be positive"); if(partitionDataQueue == null) { partitionDataQueue = new LinkedBlockingQueue(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java index 75edd840f..ecf8650c3 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-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. @@ -29,6 +29,7 @@ import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; import org.springframework.batch.core.step.JobRepositorySupport; import org.springframework.batch.core.step.StepSupport; +import org.springframework.util.StopWatch; import javax.batch.api.BatchProperty; import javax.batch.api.partition.PartitionAnalyzer; @@ -46,6 +47,7 @@ import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class JsrPartitionHandlerTests extends AbstractJsrTestCase { @@ -114,6 +116,14 @@ public class JsrPartitionHandlerTests extends AbstractJsrTestCase { handler.setJobRepository(repository); handler.afterPropertiesSet(); + + handler.setPollingInterval(-1); + try { + handler.afterPropertiesSet(); + fail("Polling interval was not checked for"); + } catch(IllegalArgumentException iae) { + assertEquals("The polling interval must be positive", iae.getMessage()); + } } @Test @@ -128,6 +138,23 @@ public class JsrPartitionHandlerTests extends AbstractJsrTestCase { assertEquals(3, count); } + @Test + public void testPollingPartitionsCompletion() throws Exception { + handler.setThreads(3); + handler.setPartitions(3); + handler.setPollingInterval(1000); + handler.afterPropertiesSet(); + + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + Collection executions = handler.handle(stepSplitter, stepExecution); + stopWatch.stop(); + + assertEquals(3, executions.size()); + assertEquals(3, count); + assertTrue(stopWatch.getLastTaskTimeMillis() >= 1000); + } + @Test public void testMapperProvidesPartitions() throws Exception { handler.setPartitionMapper(new PartitionMapper() {