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.JobState;
import com.breakersoft.plow.thrift.TaskState;

import org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean;

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)) {
            if (jobService.setJobState(job, JobState.FINISHED)) {
                logger.info("{} job has finsihed.", job.getName());
            }
            return;
        }
        dispatchPool.execute(new DispatchProcToJob(proc, job, frontEndDispatcher));
    }

}

@@ -2239,20 +2239,16 @@
         
-if (
 jobServi
@@ -2255,127 +2255,21 @@
 ce.s
-etJobState(job, JobState.FINISHED)) %7B%0A                logger.info(%22%7B%7D job has finsihed.%22, job.getName());%0A            %7D
+hutdown(job);
 %0A   

Use job shutdown function to stop job rather than just chainging the state.