Add property to configure number of messages to pop from Workflow swe…

…eper.

core/src/main/java/com/netflix/conductor/core/execution/WorkflowSweeper.java

@@ -51,21 +51,23 @@ public class WorkflowSweeper {
 
 	private int executorThreadPoolSize;
 
+	private int numberOfMessagesToPop;
+
 	private static final String className = WorkflowSweeper.class.getSimpleName();
 
 	@Inject
 	public WorkflowSweeper(WorkflowExecutor workflowExecutor, Configuration config, QueueDAO queueDAO) {
 		this.config = config;
 		this.queueDAO = queueDAO;
 		this.executorThreadPoolSize = config.getIntProperty("workflow.sweeper.thread.count", 5);
+		this.numberOfMessagesToPop = config.getIntProperty("workflow.sweeper.messages.popCount", 10);
 		if(this.executorThreadPoolSize > 0) {
 			this.executorService = Executors.newFixedThreadPool(executorThreadPoolSize);
 			init(workflowExecutor);
 			logger.info("Workflow Sweeper Initialized");
 		} else {
 			logger.warn("Workflow sweeper is DISABLED");
 		}
-
 	}
 
 	public void init(WorkflowExecutor workflowExecutor) {
@@ -77,7 +79,7 @@ public void init(WorkflowExecutor workflowExecutor) {
 					logger.info("Workflow sweep is disabled.");
 					return;
 				}
-				List<String> workflowIds = queueDAO.pop(WorkflowExecutor.DECIDER_QUEUE, 2 * executorThreadPoolSize, 2000);
+				List<String> workflowIds = queueDAO.pop(WorkflowExecutor.DECIDER_QUEUE, numberOfMessagesToPop, 2000);
 				int currentQueueSize = queueDAO.getSize(WorkflowExecutor.DECIDER_QUEUE);
 				logger.debug("Sweeper's current deciderqueue size: {}.", currentQueueSize);
 				int retrievedWorkflows = (workflowIds != null) ? workflowIds.size() : 0;