/**
 * Copyright 2016 Netflix, Inc.
 *
 * 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 com.netflix.conductor.core.execution;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.inject.Inject;
import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.netflix.conductor.core.WorkflowContext;
import com.netflix.conductor.core.config.Configuration;
import com.netflix.conductor.core.execution.ApplicationException.Code;
import com.netflix.conductor.dao.ExecutionDAO;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.metrics.Monitors;

/**
 * @author Viren
 * @author Vikram
 *
 */
@Singleton
public class WorkflowSweeper {

	private static Logger logger = LoggerFactory.getLogger(WorkflowSweeper.class);

	private ExecutorService es;
	
	protected ObjectMapper om;
	
	private DeciderService ds;
	
	private Configuration config;
	
	private QueueDAO queues;
	
	private int executorThreadPoolSize;
	
	private static final String className = WorkflowSweeper.class.getSimpleName();
		
	@Inject
	public WorkflowSweeper(ExecutionDAO edao, MetadataDAO metadata, ObjectMapper om, WorkflowExecutor workflowProvider, Configuration config, QueueDAO queues) {
		this.om = om;
		this.ds = new DeciderService(metadata, edao, om, config);
		this.config = config;
		this.queues = queues;
		this.executorThreadPoolSize = config.getIntProperty("workflow.sweeper.thread.count", 5);
		this.es = Executors.newFixedThreadPool(executorThreadPoolSize);
		init(workflowProvider);
	}

	public void init(WorkflowExecutor workflowProvider) {
		
		ScheduledExecutorService deciderPool = Executors.newScheduledThreadPool(1);
		
		deciderPool.scheduleWithFixedDelay(() -> {

			try{
				
				logger.debug("Workflow sweeping...");
				if(config.disableSweep()){
					logger.debug("Workflow sweep is disabled.");
					return;
				}
				List<String> workflowIds = queues.pop(WorkflowExecutor.deciderQueue, executorThreadPoolSize, 1000);
				sweep(workflowIds, workflowProvider);
				
			}catch(Exception e){
				Monitors.error(className, "sweep");
				logger.error(e.getMessage(), e);
				
			}
			
			
		}, 2 * config.getSweepFrequency(), 500, TimeUnit.MILLISECONDS);
	}

	public void sweep(List<String> workflowIds, WorkflowExecutor workflowProvider) throws Exception {
		
		List<Future<?>> futures = new LinkedList<>();
		for (String workflowId : workflowIds) {
			Future<?> future = es.submit(() -> {
				try {
					
					WorkflowContext ctx = new WorkflowContext(config.getAppId());
					WorkflowContext.set(ctx);
					if(logger.isDebugEnabled()) {
						logger.debug("Running sweeper for workflow {}", workflowId);
					}
					boolean done = ds.decide(workflowId, workflowProvider);
					if(!done) {
						queues.setUnackTimeout(WorkflowExecutor.deciderQueue, workflowId, config.getSweepFrequency() * 1000);
					} else {
						queues.remove(WorkflowExecutor.deciderQueue, workflowId);
					}
					
				} catch (ApplicationException e) {
					if(e.getCode().equals(Code.NOT_FOUND)) {
						logger.error("Workflow NOT found id=" + workflowId, e);
						queues.remove(WorkflowExecutor.deciderQueue, workflowId);
					}
					
				} catch (Exception e) {
					Monitors.error(className, "sweep");
					logger.error("Error running sweep for " + workflowId, e);
				}
			});
			futures.add(future);
		}

		for (Future<?> future : futures) {
			future.get();
		}
	
	}
	
}

@@ -2365,16 +2365,63 @@
 vider);%0A
+%09%09logger.info(%22Workflow Sweeper Initialized%22);%0A
 %09%7D%0A%0A%09pub
@@ -2691,37 +2691,36 @@
 ))%7B%0A%09%09%09%09%09logger.
-debug
+info
 (%22Workflow sweep
@@ -2827,16 +2827,20 @@
 erQueue,
+ 2 *
  executo
@@ -2856,17 +2856,17 @@
 olSize, 
-1
+2
 000);%0A%09%09
@@ -3038,38 +3038,11 @@
 %09%7D, 
-2 * config.getSweepFrequency()
+500
 , 50

Change to sweeper frequency logic and batch size when polling for pending workflows