BATCH-2401: fix CPU intensive loop while polling partitions completion

Before this commit, the JsrPartitionHandler was polling partitions
completion continuously. This causes a high CPU usage.

This commit makes the polling thread sleep for a configurable amount
of time in order to decrease CPU usage during the polling period.

Resolves BATCH-2401

Author: fmbenhassine

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 @@
  * 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<Serializable> partitionDataQueue;
 	private ReentrantLock lock;
@@ -72,6 +75,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
 	private JobRepository jobRepository;
 	private boolean allowStartIfComplete = false;
 	private Set<String> partitionStepNames = new HashSet<String>();
+	private int pollingInterval = DEFAULT_POLLING_INTERVAL;
 
 	/**
 	 * @return the step that will be executed by each partition
@@ -156,6 +160,14 @@ public void setJobRepository(JobRepository jobRepository) {
 		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 @@ private void processPartitionResults(
 			final List<Future<StepExecution>> tasks,
 			final Set<StepExecution> result) throws Exception {
 		while(true) {
+			Thread.sleep(pollingInterval);
 			try {
 				lock.lock();
 				while(!partitionDataQueue.isEmpty()) {
@@ -382,6 +395,7 @@ public void afterPropertiesSet() throws Exception {
 		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<Serializable>();

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.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.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 void testAfterPropertiesSet() throws Exception {
 
 		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 void testHardcodedNumberOfPartitions() throws Exception {
 		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<StepExecution> 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() {