package com.breakersoft.plow.dispatcher;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.breakersoft.plow.Task;
import com.breakersoft.plow.dispatcher.command.DispatchProcToJob;
import com.breakersoft.plow.dispatcher.domain.DispatchJob;
import com.breakersoft.plow.dispatcher.domain.DispatchProc;
import com.breakersoft.plow.event.EventManager;
import com.breakersoft.plow.rnd.thrift.RunTaskResult;
import com.breakersoft.plow.service.JobService;
import com.breakersoft.plow.thrift.TaskState;

public class BackEndDispatcher {

    private static final Logger logger =
            org.slf4j.LoggerFactory.getLogger(BackEndDispatcher.class);

    @Autowired
    EventManager eventManager;

    @Autowired
    DispatchService dispatchService;

    @Autowired
    FrontEndDispatcher frontEndDispatcher;

    @Autowired
    JobService jobService;

    private ExecutorService dispatchPool = Executors.newFixedThreadPool(8);

    public void taskComplete(RunTaskResult result) {

        Task task = jobService.getTask(result.taskId);
        DispatchJob job = frontEndDispatcher.getJob(task.getJobId());
        DispatchProc proc = dispatchService.getDispatchProc(result.procId);

        TaskState newState;
        if (result.exitStatus == 0) {
            newState =  TaskState.SUCCEEDED;
        }
        else {
            newState = TaskState.DEAD;
        }

        logger.info("{} proc reported in task completed: {}, exit status: {}",
                new Object[] { proc.getNodeName(), proc.getTaskName(), result.exitStatus });
        logger.info("New state {}", newState.toString());

        if (!jobService.stopTask(task, newState)) {
            // Task was already stopped somehow.
            // might be a retry or
            logger.warn("{} task was stopped by another thread.", proc.getTaskName());
            return;
        }
        else {
            dispatchService.unassignProc(proc);
        }

        if (!jobService.hasPendingFrames(job)) {
            jobService.shutdown(job);
            dispatchService.unbookProc(proc);
        }
        dispatchPool.execute(new DispatchProcToJob(proc, job, frontEndDispatcher));
    }

}

@@ -2117,90 +2117,174 @@
     
-jobService.shutdown(job);%0A            dispatchService.unbookProc(proc);%0A        %7D%0A
+dispatchService.unbookProc(proc);%0A            jobService.shutdown(job);%0A            logger.info(%22Job has no pending frames, unbooking%22);%0A        %7D%0A        else %7B%0A    
     
@@ -2359,21 +2359,31 @@
 spatcher));%0A
+        %7D%0A
     %7D%0A%0A%7D%0A

Make sure to unhook prods before shutting down the job.