/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.uima.ducc.orchestrator;

import java.io.File;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.uima.ducc.common.Node;
import org.apache.uima.ducc.common.NodeIdentity;
import org.apache.uima.ducc.common.SizeBytes;
import org.apache.uima.ducc.common.container.FlagsHelper;
import org.apache.uima.ducc.common.internationalization.Messages;
import org.apache.uima.ducc.common.jd.files.workitem.IRemoteLocation;
import org.apache.uima.ducc.common.jd.files.workitem.RemoteLocation;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.DuccLoggerComponents;
import org.apache.uima.ducc.common.utils.DuccPropertiesResolver;
import org.apache.uima.ducc.common.utils.DuccSchedulerClasses;
import org.apache.uima.ducc.common.utils.TimeStamp;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.orchestrator.user.UserLogging;
import org.apache.uima.ducc.orchestrator.utilities.TrackSync;
import org.apache.uima.ducc.transport.agent.IUimaPipelineAEComponent;
import org.apache.uima.ducc.transport.cmdline.ICommandLine;
import org.apache.uima.ducc.transport.cmdline.JavaCommandLine;
import org.apache.uima.ducc.transport.event.common.DuccProcess;
import org.apache.uima.ducc.transport.event.common.DuccReservation;
import org.apache.uima.ducc.transport.event.common.DuccUimaDeploymentDescriptor;
import org.apache.uima.ducc.transport.event.common.DuccWorkJob;
import org.apache.uima.ducc.transport.event.common.DuccWorkMap;
import org.apache.uima.ducc.transport.event.common.DuccWorkPopDriver;
import org.apache.uima.ducc.transport.event.common.DuccWorkReservation;
import org.apache.uima.ducc.transport.event.common.IDuccCompletionType.JobCompletionType;
import org.apache.uima.ducc.transport.event.common.IDuccCompletionType.ReservationCompletionType;
import org.apache.uima.ducc.transport.event.common.IDuccProcess;
import org.apache.uima.ducc.transport.event.common.IDuccProcessMap;
import org.apache.uima.ducc.transport.event.common.IDuccProcessType.ProcessType;
import org.apache.uima.ducc.transport.event.common.IDuccProcessWorkItems;
import org.apache.uima.ducc.transport.event.common.IDuccReservationMap;
import org.apache.uima.ducc.transport.event.common.IDuccState.JobState;
import org.apache.uima.ducc.transport.event.common.IDuccState.ReservationState;
import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType;
import org.apache.uima.ducc.transport.event.common.IDuccUimaDeploymentDescriptor;
import org.apache.uima.ducc.transport.event.common.IDuccWork;
import org.apache.uima.ducc.transport.event.common.IDuccWorkJob;
import org.apache.uima.ducc.transport.event.common.IDuccWorkReservation;
import org.apache.uima.ducc.transport.event.common.IDuccWorkService;
import org.apache.uima.ducc.transport.event.common.IProcessState.ProcessState;
import org.apache.uima.ducc.transport.event.common.IRationale;
import org.apache.uima.ducc.transport.event.common.IResourceState.ProcessDeallocationType;
import org.apache.uima.ducc.transport.event.common.IResourceState.ResourceState;
import org.apache.uima.ducc.transport.event.common.Rationale;
import org.apache.uima.ducc.transport.event.common.history.IHistoryPersistenceManager;
import org.apache.uima.ducc.transport.event.jd.IDriverStatusReport;
import org.apache.uima.ducc.transport.event.jd.IDuccProcessWorkItemsReport;
import org.apache.uima.ducc.transport.event.rm.IResource;
import org.apache.uima.ducc.transport.event.rm.IRmJobState;
import org.apache.uima.ducc.transport.event.sm.IService.ServiceState;
import org.apache.uima.ducc.transport.event.sm.ServiceDependency;
import org.apache.uima.ducc.transport.event.sm.ServiceMap;

public class StateManager {
	private static final DuccLogger logger = DuccLoggerComponents.getOrLogger(StateManager.class.getName());
	
	private static StateManager stateManager = new StateManager();
	
	public static StateManager getInstance() {
		return stateManager;
	}
	
	public StateManager() {
	}

	private OrchestratorCommonArea orchestratorCommonArea = OrchestratorCommonArea.getInstance();
	private Messages messages = orchestratorCommonArea.getSystemMessages();
	private DuccWorkMap workMap = orchestratorCommonArea.getWorkMap();
	private StateJobAccounting stateJobAccounting = StateJobAccounting.getInstance();
	
	private IHistoryPersistenceManager hpm = orchestratorCommonArea.getHistoryPersistencemanager();
	
	private boolean jobDriverTerminated(DuccWorkJob duccWorkJob) {
		String methodName = "jobDriverTerminated";
		boolean status = true;
		logger.trace(methodName, null, messages.fetch("enter"));
		IDuccProcessMap processMap = duccWorkJob.getDriver().getProcessMap();
		Iterator<DuccId> processMapIterator = processMap.keySet().iterator();
		while(processMapIterator.hasNext()) {
			DuccId duccId = processMapIterator.next();
			IDuccProcess process = processMap.get(duccId);
			if(process.isActive()) {
				logger.debug(methodName, duccId,  messages.fetch("processes active"));
				status = false;
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return status;
	}
	
	private boolean jobProcessesTerminated(DuccWorkJob duccWorkJob) {
		String methodName = "jobProcessesTerminated";
		boolean status = true;
		logger.trace(methodName, null, messages.fetch("enter"));
		IDuccProcessMap processMap = duccWorkJob.getProcessMap();
		Iterator<DuccId> processMapIterator = processMap.keySet().iterator();
		while(processMapIterator.hasNext()) {
			DuccId duccId = processMapIterator.next();
			IDuccProcess process = processMap.get(duccId);
			if(process.isActive()) {
				logger.debug(methodName, duccId,  messages.fetch("processes active"));
				status = false;
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return status;
	}
	
	private boolean allProcessesTerminated(DuccWorkJob duccWorkJob) {
		String methodName = "allProcessesTerminated";
		boolean status = false;
		logger.trace(methodName, null, messages.fetch("enter"));
		switch(duccWorkJob.getDuccType()) {
		case Job:
			if(jobDriverTerminated(duccWorkJob)) {
				if(jobProcessesTerminated(duccWorkJob)) {
					status = true;
					if(duccWorkJob.getStandardInfo().getDateOfShutdownProcessesMillis() <= 0) {
						duccWorkJob.getStandardInfo().setDateOfShutdownProcesses(TimeStamp.getCurrentMillis());
					}
				}
			}
			break;
		case Service:
			if(jobProcessesTerminated(duccWorkJob)) {
				status = true;
				if(duccWorkJob.getStandardInfo().getDateOfShutdownProcessesMillis() <= 0) {
					duccWorkJob.getStandardInfo().setDateOfShutdownProcesses(TimeStamp.getCurrentMillis());
				}
			}
			break;
		default:
			break;
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return status;
	}
	
	private long SECONDS = 1000;
	private long MINUTES = 60 * SECONDS;
	private long AgeTime = 1 * MINUTES;
	
	private boolean isAgedOut(IDuccWork duccWork) {
		String methodName = "isAgedOut";
		logger.trace(methodName, null, messages.fetch("enter"));
		boolean retVal = true;
		long endMillis = 0;
		long nowMillis = 0;
		long elapsed = 0;
		try {
			endMillis = duccWork.getStandardInfo().getDateOfCompletionMillis();
			nowMillis = System.currentTimeMillis();
			elapsed = (nowMillis - endMillis);
			if(elapsed <= AgeTime) {
				retVal = false;
			}
			endMillis = duccWork.getStandardInfo().getDateOfShutdownProcessesMillis();
			elapsed = (nowMillis - endMillis);
			if(elapsed <= AgeTime) {
				retVal = false;
			}
		}
		catch(Exception e) {
			logger.error(methodName, null, "nowMillis:"+endMillis+" "+"nowMillis:"+endMillis+" ", e);
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return retVal;
	}
	
	public boolean isSaved(IDuccWorkJob duccWorkJob) {
		String methodName = "isSaved";
		logger.trace(methodName, null, messages.fetch("enter"));
		boolean retVal = false;
		try {
			switch(duccWorkJob.getDuccType()) {
			case Job:
				hpm.saveJob(duccWorkJob);
				retVal = true;
				break;
			case Service:
				hpm.saveService((IDuccWorkService)duccWorkJob);
				retVal = true;
				break;
			default:
				break;
			}
		}
		catch(Exception e) {
			logger.error(methodName, duccWorkJob.getDuccId(), e);
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return retVal;
	}
	
	public boolean isSaved(IDuccWorkReservation duccWorkReservation) {
		String methodName = "isSaved";
		logger.trace(methodName, null, messages.fetch("enter"));
		boolean retVal = false;
		try {
			hpm.saveReservation(duccWorkReservation);
			retVal = true;
		}
		catch(Exception e) {
			e.printStackTrace();
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return retVal;
	}
	
	public int prune(DuccWorkMap workMap) {
		String methodName = "prune";
		int changes = 0;
		logger.trace(methodName, null, messages.fetch("enter"));
		long t0 = System.currentTimeMillis();
		Iterator<DuccId> workMapIterator = workMap.keySet().iterator();
		while(workMapIterator.hasNext()) {
			DuccId duccId = workMapIterator.next();
			IDuccWork duccWork = WorkMapHelper.findDuccWork(workMap, duccId, this, methodName);
			if(duccWork == null) {
				logger.warn(methodName, duccId, "not found");
				continue;
			}
			DuccType duccType = duccWork.getDuccType();
			if(duccType == null) {
				logger.warn(methodName, duccId, "no type?");
				continue;
			}
			switch(duccType) {
			case Job:
			case Service:
				DuccWorkJob duccWorkJob = (DuccWorkJob)duccWork;
				if(duccWorkJob != null) {
					if(duccWorkJob.isCompleting() && allProcessesTerminated(duccWorkJob)) {
						stateJobAccounting.stateChange(duccWorkJob, JobState.Completed);
					}
					if(duccWorkJob.isCompleted() && allProcessesTerminated(duccWorkJob) && isSaved(duccWorkJob) && isAgedOut(duccWorkJob)) {
						WorkMapHelper.removeDuccWork(workMap, duccWorkJob, this, methodName);
						logger.info(methodName, duccId, messages.fetch("removed job"));
						changes ++;
						IDuccProcessMap processMap = duccWorkJob.getProcessMap();
						Iterator<DuccId> processMapIterator = processMap.keySet().iterator();
						while(processMapIterator.hasNext()) {
							DuccId processDuccId = processMapIterator.next();
							orchestratorCommonArea.getProcessAccounting().removeProcess(processDuccId);
							logger.info(methodName, duccId, messages.fetch("removed process")+" "+processDuccId.toString());
							changes ++;
						}
						logger.info(methodName, duccId, messages.fetch("processes inactive"));
					}
					else {
						logger.debug(methodName, duccId, messages.fetch("processes active"));
					}
				}
				break;
			case Reservation:
				DuccWorkReservation duccWorkReservation = (DuccWorkReservation)duccWork;
				if(duccWorkReservation != null) {
					if(duccWorkReservation.isCompleted() && isSaved(duccWorkReservation) && isAgedOut(duccWorkReservation)) {
						WorkMapHelper.removeDuccWork(workMap, duccWorkReservation, this, methodName);
						logger.info(methodName, duccId, messages.fetch("removed reservation"));
						changes ++;
					}
				}
				break;
			default:
				break;
			}
		}
		long t1 = System.currentTimeMillis();
		long elapsed = t1 - t0;
		if(elapsed > Constants.SYNC_LIMIT) {
			logger.debug(methodName, null, "elapsed msecs: "+elapsed);
		}
		logger.debug(methodName, null, "processToWorkMap.size()="+orchestratorCommonArea.getProcessAccounting().processCount());
		if(changes > 0) {
			OrchestratorCheckpoint.getInstance().saveState();
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return changes;
	}
	
	private int stateChange(DuccWorkJob duccWorkJob, JobState state) {
		stateJobAccounting.stateChange(duccWorkJob, state);
		return 1;
	}
	
	private int stateChange(DuccWorkReservation duccWorkReservation, ReservationState state) {
		duccWorkReservation.stateChange(state);
		return 1;
	}
	
	private void setJdJmxUrl(DuccWorkJob job, String jdJmxUrl) {
		if(jdJmxUrl != null) {
			DuccWorkPopDriver driver = job.getDriver();
			IDuccProcessMap processMap = driver.getProcessMap();
			if(processMap != null) {
				Collection<IDuccProcess> processCollection = processMap.values();
				Iterator<IDuccProcess> iterator = processCollection.iterator();
				while(iterator.hasNext()) {
					IDuccProcess process = iterator.next();
					process.setProcessJmxUrl(jdJmxUrl);
				}
			}
		}
	}
	
	private void copyInvestmentReport(DuccWorkJob job, IDriverStatusReport jdStatusReport) {
		String methodName = "copyInvestmentReport";
		try {
			Map<RemoteLocation, Long> omMap = jdStatusReport.getInvestmentMillisMap();
			IDuccProcessMap processMap = job.getProcessMap();
			for(Entry<DuccId, IDuccProcess> entry : processMap.entrySet()) {
				IDuccProcess process = entry.getValue();
				Node node = process.getNode();
				NodeIdentity nodeIdentity = node.getNodeIdentity();
				String nodeIP = nodeIdentity.getIp();
				String pid = process.getPID();
				RemoteLocation remoteLocation = new RemoteLocation(nodeIP, pid);
				long investment = 0;
				if(omMap.containsKey(remoteLocation)) {
					investment = omMap.get(remoteLocation).longValue();
				}
				process.setWiMillisInvestment(investment);
				logger.debug(methodName, job.getDuccId(), process.getDuccId(), "investment:"+investment+" "+"node(IP): "+nodeIP+" "+"pid: "+pid);
			}
		}
		catch(Exception e) {
			logger.error(methodName, job.getDuccId(), e);
		}
	}
	private void copyProcessWorkItemsReport(DuccWorkJob job, IDriverStatusReport jdStatusReport) {
		String methodName = "copyProcessWorkItemsReport";	
		try {
			IDuccProcessMap processMap = job.getProcessMap();
			IDuccProcessWorkItemsReport pwiReport = jdStatusReport.getDuccProcessWorkItemsMap();
			if(pwiReport!= null) {
				ConcurrentHashMap<DuccId, IDuccProcessWorkItems> pwiMap = pwiReport.getMap();
				Iterator<DuccId> iterator = pwiMap.keySet().iterator();
				while(iterator.hasNext()) {
					DuccId processId = iterator.next();
					IDuccProcess process = processMap.get(processId);
					IDuccProcessWorkItems pwi = pwiMap.get(processId);
					process.setProcessWorkItems(pwi);
					logger.trace(methodName, job.getDuccId(), "done:"+pwi.getCountDone()+" "+"error:"+pwi.getCountError()+" "+"dispatch:"+pwi.getCountDispatch());
				}
			}
		}
		catch(Exception e) {
			logger.error(methodName, job.getDuccId(), e);
		}
	}
	
	private void copyDriverWorkItemsReport(DuccWorkJob job, IDriverStatusReport jdStatusReport) {
		String methodName = "copyDriverWorkItemsReport";
		try {
			IDuccProcessWorkItemsReport pwiReport = jdStatusReport.getDuccProcessWorkItemsMap();
			if(pwiReport != null) {
				IDuccProcessWorkItems pwi = pwiReport.getTotals();
				DuccWorkPopDriver driver = job.getDriver();
				IDuccProcessMap processMap = driver.getProcessMap();
				if(processMap != null) {
					Iterator<DuccId> iterator = processMap.keySet().iterator();
					while(iterator.hasNext()) {
						DuccId processId = iterator.next();
						IDuccProcess process = processMap.get(processId);
						process.setProcessWorkItems(pwi);
						logger.debug(methodName, job.getDuccId(), "done:"+pwi.getCountDone()+" "+"error:"+pwi.getCountError()+" "+"dispatch:"+pwi.getCountDispatch());
					}
				}
			}
			job.setWiMillisMin(jdStatusReport.getWiMillisMin());
			job.setWiMillisMax(jdStatusReport.getWiMillisMax());
			job.setWiMillisAvg(jdStatusReport.getWiMillisAvg());
			job.setWiMillisOperatingLeast(jdStatusReport.getWiMillisOperatingLeast());
			job.setWiMillisCompletedMost(jdStatusReport.getWiMillisCompletedMost());
		}
		catch(Exception e) {
			logger.error(methodName, job.getDuccId(), e);
		}
	}
	
	private void setCompletionIfNotAlreadySet(DuccWorkJob duccWorkJob, IDriverStatusReport jdStatusReport) {
		String methodName = "setCompletionIfNotAlreadySet";
		DuccId jobid = null;
		try {
			jobid = duccWorkJob.getDuccId();
			setCompletionIfNotAlreadySet(jobid, duccWorkJob, jdStatusReport.getJobCompletionType(), jdStatusReport.getJobCompletionRationale());
		}
		catch(Exception e) {
			logger.error(methodName, jobid, e);
		}
	}
	
	private void setCompletionIfNotAlreadySet(DuccWorkJob duccWorkJob, JobCompletionType jobCompletionType, IRationale rationale) {
		String methodName = "setCompletionIfNotAlreadySet";
		DuccId jobid = null;
		try {
			jobid = duccWorkJob.getDuccId();
			setCompletionIfNotAlreadySet(jobid, duccWorkJob, jobCompletionType,rationale);
		}
		catch(Exception e) {
			logger.error(methodName, jobid, e);
		}
	}
	
	private void setCompletionIfNotAlreadySet(DuccId jobid, DuccWorkJob duccWorkJob, JobCompletionType reqJobCompletionType, IRationale reqRationale) {
		String methodName = "setCompletionIfNotAlreadySet";
		logger.trace(methodName, null, messages.fetch("enter"));
		try {
			JobCompletionType curJobCompletionType = duccWorkJob.getCompletionType();
			switch(curJobCompletionType) {
			case Undefined:
				duccWorkJob.setCompletion(reqJobCompletionType, reqRationale);
				logger.debug(methodName, jobid, "changed: "+curJobCompletionType+" to "+reqJobCompletionType);
				break;
			default:
				logger.debug(methodName, jobid, "unchanged: "+curJobCompletionType+" to "+reqJobCompletionType);
				break;
			}
			
		}
		catch(Exception e) {
			logger.error(methodName, jobid, e);
		}
		logger.trace(methodName, null, messages.fetch("exit"));
	}
	
	private void addJdUrlToJpCommandLine(IDuccWorkJob dwj, IDriverStatusReport jdStatusReport) {
		String location = "addJdPortToJpCommandLine";
		DuccId jobid = null;
		if(!dwj.isJdURLSpecified()) {
			jobid = dwj.getDuccId();
			String node = jdStatusReport.getNode();
			int port = jdStatusReport.getPort();
			if(port > 0) {
				JavaCommandLine jcl = (JavaCommandLine) dwj.getCommandLine();
				// add port
				String opt;
				// format is http://<node>:<port>/jdApp
				String value = "http://"+node+":"+port+"/jdApp";
				opt = FlagsHelper.Name.JdURL.dname()+"="+value;
				jcl.addOption(opt);
				logger.info(location, jobid, opt);
				dwj.setJdURLSpecified();
			}
		}
	}
	
	private void addDeployableToJpCommandLine(IDuccWorkJob dwj, IDriverStatusReport jdStatusReport) {
		String location = "addDeployableToJpCommandLine";
		DuccId jobid = null;
		if(!dwj.isDdSpecified()) {
			//V1
			String jpDd = jdStatusReport.getUimaDeploymentDescriptor();
			if(jpDd != null) {
				IDuccUimaDeploymentDescriptor uimaDeploymentDescriptor = new DuccUimaDeploymentDescriptor(jpDd);
				dwj.setUimaDeployableConfiguration(uimaDeploymentDescriptor);
			}
			//V2
			boolean agentAddsDdToCommandLine = false;
			if(!agentAddsDdToCommandLine) {
				String jpAe = jdStatusReport.getUimaAnalysisEngine();
				if(jpAe != null) {
					ICommandLine jcl = dwj.getCommandLine();
					List<String> args = jcl.getArguments();
					String arg = jpAe;
					if(args == null) {
						jcl.addArgument(arg);
						logger.debug(location, jobid,  "add[null]:"+arg);
					}
					else if(args.isEmpty()) {
						jcl.addArgument(jpAe);
						logger.debug(location, jobid, "add[empty]:"+arg);
					}
					List<String> argList = jcl.getArguments();
					if(args != null) {
						int index = 0;
						for(String argument : argList) {
							logger.debug(location, jobid, "arg["+index+"]: "+argument);
							index++;
						}
					}
					dwj.setDdSpecified();
				}
			}
		}
	}
	
	/**
	 * JD reconciliation
	 */
	public void reconcileState(IDriverStatusReport jdStatusReport) {
		String methodName = "reconcileState (JD)";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		TrackSync ts = TrackSync.await(workMap, this.getClass(), methodName);
		synchronized(workMap) {
			ts.using();
			DuccId duccId = jdStatusReport.getDuccId();
			String sid = ""+duccId.getFriendly();
			DuccWorkJob duccWorkJob = (DuccWorkJob) WorkMapHelper.findDuccWork(workMap, sid, this, methodName);
			if(duccWorkJob != null) {
				//
				String jdJmxUrl = jdStatusReport.getJdJmxUrl();
				setJdJmxUrl(duccWorkJob, jdJmxUrl);
				//
				copyInvestmentReport(duccWorkJob, jdStatusReport);
				copyProcessWorkItemsReport(duccWorkJob, jdStatusReport);
				copyDriverWorkItemsReport(duccWorkJob, jdStatusReport);
				//
				switch(duccWorkJob.getJobState()) {
				case Completed:
					break;
				case Completing:
				default:
					duccWorkJob.setWiTotal(jdStatusReport.getWorkItemsTotal());
					duccWorkJob.setWiDone(jdStatusReport.getWorkItemsProcessingCompleted());
					duccWorkJob.setWiError(jdStatusReport.getWorkItemsProcessingError());
					break;
				}
				//
				IRationale rationale;
				if(jdStatusReport.getWorkItemsTotal() == 0) {
					jobTerminate(duccWorkJob, JobCompletionType.NoWorkItemsFound, new Rationale("job driver had no work items to process"), ProcessDeallocationType.JobCanceled);
				}
				else if(jdStatusReport.isKillJob()) {
					rationale = jdStatusReport.getJobCompletionRationale();
					jobTerminate(duccWorkJob, JobCompletionType.CanceledByDriver, rationale, ProcessDeallocationType.JobFailure);
				}
				else {
					switch(jdStatusReport.getDriverState()) {
					case Failed:
						rationale = jdStatusReport.getJobCompletionRationale();
						jobTerminate(duccWorkJob, JobCompletionType.CanceledByDriver, rationale, ProcessDeallocationType.JobFailure);
						break;
					case NotRunning:
						break;
					case Initializing:	
						switch(duccWorkJob.getJobState()) {
						case WaitingForDriver: 
						    addJdUrlToJpCommandLine(duccWorkJob, jdStatusReport);
						    addDeployableToJpCommandLine(duccWorkJob, jdStatusReport);
						    if(!duccWorkJob.isJdURLSpecified()) {
						        logger.debug(methodName, duccId, "No JdURL provided yet - still waitingForDriver");
						        break;
	                        }
							JobState nextState = JobState.WaitingForServices;
							if(duccWorkJob.getServiceDependencies() == null) {
								String message = messages.fetch("bypass")+" "+nextState;
								logger.debug(methodName, duccId, message);
								nextState = JobState.WaitingForResources;
							}
							stateJobAccounting.stateChange(duccWorkJob, nextState);
							break;
						case Assigned:	
							stateJobAccounting.stateChange(duccWorkJob, JobState.Initializing);
							break;
						case Initializing:
							break;
						default:
							break;
						}
						break;
					case Running:
					case Idle:	
						if(jdStatusReport.isKillJob()) {
							rationale = jdStatusReport.getJobCompletionRationale();
							switch(duccWorkJob.getJobState()) {
							case WaitingForServices:
								if(rationale == null) {
									rationale = new Rationale("waiting for services");
								}
								else {
									if(rationale.isSpecified()) {
										String text = rationale.getText();
										rationale = new Rationale(text+": "+"waiting for services");
									}
									else {
										rationale = new Rationale("waiting for services");
									}
								}
								break;
							default:
								break;
							}
							jobTerminate(duccWorkJob, JobCompletionType.CanceledByDriver, rationale, ProcessDeallocationType.JobFailure);
							break;
						}
						switch(duccWorkJob.getJobState()) {
						case WaitingForDriver:
							stateJobAccounting.stateChange(duccWorkJob, JobState.WaitingForServices);
							break;
						case Assigned:	
						case Initializing:
							stateJobAccounting.stateChange(duccWorkJob, JobState.Running);
							break;
						default:
							break;
						}
						break;
					case Completing:	
						if(!duccWorkJob.isFinished()) {
							stateJobAccounting.stateChange(duccWorkJob, JobState.Completing);
						}
						stopJps(duccWorkJob);
						stopJd(duccWorkJob);
						break;
					case Completed:
						if(!duccWorkJob.isCompleted()) {
							if(!duccWorkJob.isFinished()) {
								stateJobAccounting.stateChange(duccWorkJob, JobState.Completing);
							}
							stopJps(duccWorkJob);
							stopJd(duccWorkJob);
							duccWorkJob.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
							switch(jdStatusReport.getJobCompletionType()) {
							case EndOfJob:
								try {
									int errors = jdStatusReport.getWorkItemsProcessingError();
									int done = jdStatusReport.getWorkItemsProcessingCompleted();
									if(errors > 0) {
										setCompletionIfNotAlreadySet(duccWorkJob, JobCompletionType.Error, new Rationale("state manager detected error work items="+errors));
									}
									else if(done == 0) {
										setCompletionIfNotAlreadySet(duccWorkJob, JobCompletionType.EndOfJob, new Rationale("state manager detected no work items processed"));
									}
									else {
										setCompletionIfNotAlreadySet(duccWorkJob, JobCompletionType.EndOfJob, new Rationale("state manager detected normal completion"));
									}
								}
								catch(Exception e) {
									logger.error(methodName, duccId, e);
								}
								break;
							default:
								setCompletionIfNotAlreadySet(duccWorkJob, jdStatusReport);
								break;
							}
						}
						break;
					case Undefined:
						break;
					}
				}
				//
				OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(jdStatusReport,duccWorkJob);
				if(deallocateIdleProcesses(duccWorkJob, jdStatusReport)) {
					changes++;
				}
				if(deallocateFailedProcesses(duccWorkJob, jdStatusReport)) {
					changes++;
				}
			}
			else {
				logger.warn(methodName, duccId, messages.fetch("not found"));
			}
		}
		ts.ended();
		if(changes > 0) {
			OrchestratorCheckpoint.getInstance().saveState();
		}
		logger.trace(methodName, null, messages.fetch("exit"));
	}
	
	private boolean isExcessCapacity(DuccWorkJob job) {
		String methodName = "isExcessCapacity";
		boolean retVal = false;
		long total = job.getWiTotal();
		long done = job.getWiDone();
		long error = job.getWiError();
		long sum = total + done + error;
		if(sum > 0) {
			long capacity = job.getWorkItemCapacity();
			long todo = total - (done + error);
			long tpp = job.getSchedulingInfo().getIntThreadsPerProcess();
			long numShares = 0;
			if(todo%tpp > 0) {
				numShares = 1;
			}
			numShares += todo / tpp;
			long adjTodo = numShares * tpp;
			if(capacity > 0) {
				if(adjTodo < capacity) {
					retVal = true;
				}
			}
			logger.info(methodName, job.getDuccId(), "todo:"+todo+" "+"adjTodo:"+adjTodo+" "+"capacity:"+capacity+" "+"excess:"+retVal);
		}
		else {
			logger.info(methodName, job.getDuccId(), "todo:"+"?"+" "+"capacity:"+"?"+" "+"excess:"+retVal);
		}
		return retVal;
	}
	
	private boolean isDeallocatable(IDriverStatusReport jdStatusReport) {
		boolean retVal = false;
		if(!jdStatusReport.isPending()) {
			retVal = true;
		}
		return retVal;
	}
	
	private boolean deallocateIdleProcesses(DuccWorkJob job, IDriverStatusReport jdStatusReport) {
		String methodName = "deallocateIdleProcesses";
		boolean retVal = false;
		DuccId jobid = job.getDuccId();
		try {
			if(isDeallocatable(jdStatusReport)) {
				IDuccProcessMap processMap = job.getProcessMap();
				Iterator<DuccId> iterator = processMap.keySet().iterator();
				boolean excessCapacity = isExcessCapacity(job);
				int count = 0;
				while(iterator.hasNext() && excessCapacity) {
					count++;
					DuccId duccId = iterator.next();
					IDuccProcess process = processMap.get(duccId);
					if(!process.isDeallocated()) {
						String nodeIP = process.getNodeIdentity().getIp();
						String PID = process.getPID();
						if(!jdStatusReport.isOperating(nodeIP, PID)) {
							OrUtil.setResourceState(job, process, ResourceState.Deallocated);
							process.setProcessDeallocationType(ProcessDeallocationType.Voluntary);
							logger.info(methodName, job.getDuccId(), process.getDuccId(), "deallocated");
							retVal = true;
							excessCapacity = isExcessCapacity(job);
						}
						else {
							logger.debug(methodName, job.getDuccId(), process.getDuccId(), "operating");
						}
					}
					else {
						logger.trace(methodName, job.getDuccId(), process.getDuccId(), "already deallocated");
					}
				}
				logger.trace(methodName, jobid, "count="+count);
			}
			else {
				logger.debug(methodName, jobid, "not deallocatable");
			}
		}
		catch(Exception e) {
			logger.error(methodName, job.getDuccId(), e);
		}
		return retVal;
	}
	
	private DuccId getDuccId(IDuccProcessMap processMap, IRemoteLocation remoteLocation) {
		String methodName = "getDuccId";
		DuccId duccId = null;
		if(processMap != null) {
			if(remoteLocation != null) {
				String nodeIp = remoteLocation.getNodeIP();
				String processId = remoteLocation.getPid();
				logger.debug(methodName, null, "nodeIp:"+nodeIp+" "+"processId:"+processId);
				for(Entry<DuccId, IDuccProcess> entry : processMap.entrySet()) {
					IDuccProcess p = entry.getValue();
					logger.debug(methodName, null, "duccId:"+entry.getKey()+" "+"nodeId:"+p.getNodeIdentity().getIp()+" "+"processId:"+p.getPID());
				}
				IDuccProcess process = processMap.findProcess(nodeIp, processId);
				if(process != null) {
					duccId = process.getDuccId();
				}
			}
			else {
				logger.debug(methodName, null, "remoteLocation null?");
			}
		}
		else {
			logger.debug(methodName, null, "processMap null?");
		}
		return duccId;
	}
	
	private boolean deallocateFailedProcesses(DuccWorkJob job, IDriverStatusReport jdStatusReport) {
		String methodName = "deallocateFailedProcesses";
		boolean retVal = false;
		IDuccProcessMap processMap = job.getProcessMap();
		Map<IRemoteLocation, ProcessDeallocationType> map = jdStatusReport.getProcessKillMap();
		if(map != null) {
			logger.debug(methodName, job.getDuccId(), "size:"+map.size());
			for(Entry<IRemoteLocation, ProcessDeallocationType> entry : map.entrySet()) {
				DuccId duccId = getDuccId(processMap, entry.getKey());
				if(duccId != null) {
					IDuccProcess process = processMap.get(duccId);
					if(process != null) {
						if(!process.isDeallocated()) {
							OrUtil.setResourceState(job, process, ResourceState.Deallocated);
							process.setProcessDeallocationType(entry.getValue());
							logger.info(methodName, job.getDuccId(), process.getDuccId(), "deallocated");
						}
					}
					else {
						logger.warn(methodName, job.getDuccId(), duccId, "not in process map");
					}
				}
				else {
					logger.warn(methodName, job.getDuccId(), duccId, "null?");
				}
			}
		}
		return retVal;
	}
	
	private void stopJps(DuccWorkJob job) {
		String methodName = "stopJps";
		IDuccProcessMap processMap = job.getProcessMap();
		Iterator<DuccId> iterator = processMap.keySet().iterator();
		while (iterator.hasNext()) {
			DuccId duccId = iterator.next();
			IDuccProcess process = processMap.get(duccId);
			if(process != null) {
				if(!process.isDeallocated()) {
					OrUtil.setResourceState(job, process, ResourceState.Deallocated);
					process.setProcessDeallocationType(ProcessDeallocationType.Voluntary);
					logger.info(methodName, job.getDuccId(), process.getDuccId(), "deallocated");
				}
			}
		}
	}
	
	private boolean delayStopJd(DuccWorkJob job) {
		String methodName = "delayStopJd";
		boolean retVal = true;
		job.setCompletingTOD();
		if(!job.hasAliveProcess()) {
			long elapsed = System.currentTimeMillis() - job.getCompletingTOD();
			if((System.currentTimeMillis() - job.getCompletingTOD()) > 1000*60*5) {
				logger.debug(methodName, job.getDuccId(), ""+elapsed);
				retVal = false;
			}
			else if(!job.hasNoPidProcess()) {
				logger.debug(methodName, job.getDuccId(), "all JPs have PIDs");
				retVal = false;
			}
		}
		if(retVal) {
			logger.debug(methodName, job.getDuccId(), "");
		}
		return retVal;
	}
	
	private void stopJd(DuccWorkJob job) {
		String methodName = "stopJd";
		if(!delayStopJd(job)) {
			IDuccProcessMap processMap = job.getDriver().getProcessMap();
			Iterator<DuccId> iterator = processMap.keySet().iterator();
			while (iterator.hasNext()) {
				DuccId duccId = iterator.next();
				IDuccProcess process = processMap.get(duccId);
				if(process != null) {
					if(!process.isDeallocated()) {
						OrUtil.setResourceState(job, process, ResourceState.Deallocated);
						process.setProcessDeallocationType(ProcessDeallocationType.Voluntary);
						logger.info(methodName, job.getDuccId(), process.getDuccId(), "deallocated");
					}
				}
			}
		}
	}
	
	private static AtomicBoolean refusedLogged = new AtomicBoolean(false);
	
	/**
	 * RM reconciliation
	 */
	public void reconcileState(Map<DuccId, IRmJobState> rmResourceStateMap) throws Exception {
		String methodName = "reconcileState (RM)";
		logger.trace(methodName, null, messages.fetch("enter"));
		logger.debug(methodName, null, messages.fetchLabel("size")+rmResourceStateMap.size());
		int changes = 0;
		TrackSync ts = TrackSync.await(workMap, this.getClass(), methodName);
		synchronized(workMap) {
			ts.using();
			Iterator<DuccId> rmResourceStateIterator = rmResourceStateMap.keySet().iterator();
			while(rmResourceStateIterator.hasNext()) {
				DuccId duccId = rmResourceStateIterator.next();
				IRmJobState rmResourceState = rmResourceStateMap.get(duccId);
				Map<DuccId, IResource> mapAdditions = rmResourceState.getPendingAdditions();
				if(mapAdditions != null) {
					int mapSize = mapAdditions.size();
					if(mapSize > 0) {
						logger.info(methodName, duccId, messages.fetchLabel("pending additions")+mapSize);
					}
					else {
						logger.trace(methodName, duccId, messages.fetchLabel("pending additions")+mapSize);
					}
					
				}
				Map<DuccId, IResource> mapRemovals = rmResourceState.getPendingRemovals();
				if(mapRemovals != null) {
					int mapSize = mapRemovals.size();
					if(mapSize > 0) {
						logger.info(methodName, duccId, messages.fetchLabel("pending removals")+mapSize);
					}
					else {
						logger.trace(methodName, duccId, messages.fetchLabel("pending removals")+mapSize);
					}
				}
				IDuccWork duccWork = WorkMapHelper.findDuccWork(workMap, duccId, this, methodName);
				if(duccWork== null) {
					logger.debug(methodName, duccId, messages.fetch("not found"));
				}
				else {
					logger.trace(methodName, duccId, messages.fetchLabel("type")+duccWork.getDuccType());
					duccWork.setRmReason(null);
					switch(duccWork.getDuccType()) {
					case Job:
						logger.trace(methodName, duccId, messages.fetch("processing job..."));
						DuccWorkJob duccWorkJob = (DuccWorkJob) duccWork;
						processPurger(duccWorkJob,rmResourceState.getResources());
						changes += processMapResourcesAdd(duccWorkJob,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingAdditions());
						changes += processMapResourcesDel(duccWorkJob,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingRemovals());
						JobState jobState = duccWorkJob.getJobState();
						logger.trace(methodName, duccId, messages.fetchLabel("job state")+jobState);
						switch(jobState) {
						case Received:
						case WaitingForDriver:
							logger.warn(methodName, duccId, messages.fetchLabel("unexpected state")+jobState);
							break;
						case WaitingForServices:
							logger.debug(methodName, duccId, messages.fetchLabel("unexpected state")+jobState);
							break;
						case WaitingForResources:
							String rmReason = rmResourceState.getReason();
							logger.trace(methodName, duccId, messages.fetchLabel("rmReason")+rmReason);
							if(rmResourceState.isRefused()) {
								duccWorkJob.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
								duccWorkJob.setCompletionType(JobCompletionType.ResourcesUnavailable);
								duccWorkJob.setCompletionRationale(new Rationale("resource manager refused allocation: "+rmResourceState.getReason()));
								changes += stateChange(duccWorkJob,JobState.Completed);
								logger.warn(methodName, duccId, messages.fetchLabel("refused")+rmResourceState.getReason());
								String userName = duccWorkJob.getStandardInfo().getUser();
								String userLogDir = duccWorkJob.getUserLogsDir()+duccWorkJob.getDuccId().getFriendly()+File.separator;;
								String text = rmResourceState.getReason();
								UserLogging.record(userName, userLogDir, text);
							}
							if(duccWorkJob.getProcessMap().size() > 0) {
								changes += stateChange(duccWorkJob,JobState.Assigned);
								logger.info(methodName, duccId, messages.fetchLabel("resources count")+duccWorkJob.getProcessMap().size());
							}
							break;
						case Assigned:	
						case Initializing:
						case Running:
							if(duccWorkJob.getProcessMap().size() == 0) {
								changes += stateChange(duccWorkJob,JobState.WaitingForResources);
								logger.info(methodName, duccId, messages.fetchLabel("resources count")+duccWorkJob.getProcessMap().size());
							}
							break;
						case Completing:
						case Completed:
							logger.debug(methodName, duccId, messages.fetchLabel("unsuitable state")+jobState);
							break;
						case Undefined:
							logger.warn(methodName, duccId, messages.fetchLabel("unsuitable state")+jobState);
							break;
						}
						break;
					case Reservation:
						logger.trace(methodName, duccId, messages.fetch("processing reservation..."));
						DuccWorkReservation duccWorkReservation = (DuccWorkReservation) duccWork;
						changes += reservationMapResourcesAdd(duccWorkReservation,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingAdditions());
						changes += reservationMapResourcesDel(duccWorkReservation,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingRemovals());
						ReservationState reservationState = duccWorkReservation.getReservationState();
						logger.trace(methodName, duccId, messages.fetchLabel("reservation state")+reservationState);
						switch(reservationState) {
						case Received:
							logger.warn(methodName, duccId, messages.fetchLabel("unexpected state")+reservationState);
							break;
						case WaitingForResources:
							String rmReason = rmResourceState.getReason();
							logger.trace(methodName, duccId, messages.fetchLabel("rmReason")+rmReason);
							duccWork.setRmReason(rmReason);
							if(rmResourceState.isRefused()) {
								String schedulingClass = duccWorkReservation.getSchedulingInfo().getSchedulingClass().trim();
								if(schedulingClass.equals(DuccSchedulerClasses.JobDriver)) {
									if(!refusedLogged.get()) {
										logger.warn(methodName, duccId, messages.fetchLabel("refusal ignored")+rmResourceState.getReason());
										refusedLogged.set(true);
									}
								}
								else {
									duccWorkReservation.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
									duccWorkReservation.setCompletionType(ReservationCompletionType.ResourcesUnavailable);
									duccWorkReservation.setCompletionRationale(new Rationale("resource manager refused allocation: "+rmResourceState.getReason()));
									changes += stateChange(duccWorkReservation,ReservationState.Completed);
									logger.warn(methodName, duccId, messages.fetchLabel("refused")+rmResourceState.getReason());
								}
							}
							else {
								if(rmResourceState.getResources() != null) {
									if(!rmResourceState.getResources().isEmpty()) {
										changes += stateChange(duccWorkReservation,ReservationState.Assigned);
										logger.info(methodName, duccId, messages.fetchLabel("resources count")+rmResourceState.getResources().size());
									}
								}
								else {
									logger.info(methodName, duccId, messages.fetch("waiting...no resources?"));
								}
							}
							break;
						case Assigned:
							if(rmResourceState.getResources() != null) {
								if(rmResourceState.getResources().isEmpty()) {
									changes += stateChange(duccWorkReservation,ReservationState.Completed);
									logger.info(methodName, duccId, messages.fetchLabel("resources count")+rmResourceState.getResources().size());
								}
							}
							else {
								logger.info(methodName, duccId, messages.fetch("assigned...no resources?"));
							}
							break;
						case Completed:
							logger.debug(methodName, duccId, messages.fetchLabel("unsuitable state")+reservationState);
							break;
						case Undefined:
							logger.warn(methodName, duccId, messages.fetchLabel("unsuitable state")+reservationState);
							break;
						}
						break;
					case Service:
						logger.trace(methodName, duccId, messages.fetch("processing service..."));
						DuccWorkJob duccWorkService = (DuccWorkJob) duccWork;
						int processPurged = processPurger(duccWorkService,rmResourceState.getResources());
						changes += processMapResourcesAdd(duccWorkService,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingAdditions());
						changes += processMapResourcesDel(duccWorkService,rmResourceState.memoryGbPerProcess(),rmResourceState.getPendingRemovals());
						JobState serviceState = duccWorkService.getJobState();
						logger.trace(methodName, duccId, messages.fetchLabel("service state")+serviceState);
						switch(serviceState) {
						case Received:
							logger.warn(methodName, duccId, messages.fetchLabel("unexpected state")+serviceState);
							break;
						case WaitingForServices:
							logger.debug(methodName, duccId, messages.fetchLabel("unexpected state")+serviceState);
							break;
						case WaitingForResources:
							String rmReason = rmResourceState.getReason();
							logger.trace(methodName, duccId, messages.fetchLabel("rmReason")+rmReason);
							duccWork.setRmReason(rmReason);
							if(rmResourceState.isRefused()) {
								duccWorkService.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
								duccWorkService.setCompletionType(JobCompletionType.ResourcesUnavailable);
								duccWorkService.setCompletionRationale(new Rationale("resource manager refused allocation: "+rmResourceState.getReason()));
								changes += stateChange(duccWorkService,JobState.Completed);
								logger.warn(methodName, duccId, messages.fetchLabel("refused")+rmResourceState.getReason());
								String userName = duccWorkService.getStandardInfo().getUser();
								String userLogDir = duccWorkService.getUserLogsDir()+duccWorkService.getDuccId().getFriendly()+File.separator;;
								String text = rmResourceState.getReason();
								UserLogging.record(userName, userLogDir, text);
							}
							if(duccWorkService.getProcessMap().size() > 0) {
								changes += stateChange(duccWorkService,JobState.Initializing);
								logger.info(methodName, duccId, messages.fetchLabel("resources count")+duccWorkService.getProcessMap().size());
							}
							if((processPurged > 0) && allProcessesTerminated(duccWorkService)) {
								duccWorkService.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
								duccWorkService.setCompletionType(JobCompletionType.ResourcesUnavailable);
								duccWorkService.setCompletionRationale(new Rationale("resource manager purged allocation: "+rmResourceState.getReason()));
								changes += stateChange(duccWorkService,JobState.Completed);
								logger.warn(methodName, duccId, messages.fetchLabel("purged")+rmResourceState.getReason());
								String userName = duccWorkService.getStandardInfo().getUser();
								String userLogDir = duccWorkService.getUserLogsDir()+duccWorkService.getDuccId().getFriendly()+File.separator;;
								String text = rmResourceState.getReason();
								UserLogging.record(userName, userLogDir, text);
							}
							break;
						case Assigned:	
						case Initializing:
						case Running:
							if(duccWorkService.getProcessMap().size() == 0) {
								changes += stateChange(duccWorkService,JobState.WaitingForResources);
								logger.info(methodName, duccId, messages.fetchLabel("resources count")+duccWorkService.getProcessMap().size());
							}
							if((processPurged > 0) && allProcessesTerminated(duccWorkService)) {
								duccWorkService.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
								duccWorkService.setCompletionType(JobCompletionType.ResourcesUnavailable);
								duccWorkService.setCompletionRationale(new Rationale("resource manager purged allocation: "+rmResourceState.getReason()));
								changes += stateChange(duccWorkService,JobState.Completed);
								logger.warn(methodName, duccId, messages.fetchLabel("purged")+rmResourceState.getReason());
								String userName = duccWorkService.getStandardInfo().getUser();
								String userLogDir = duccWorkService.getUserLogsDir()+duccWorkService.getDuccId().getFriendly()+File.separator;;
								String text = rmResourceState.getReason();
								UserLogging.record(userName, userLogDir, text);
							}
							break;
						case Completing:
						case Completed:
							logger.debug(methodName, duccId, messages.fetchLabel("unsuitable state")+serviceState);
							break;
						case Undefined:
							logger.warn(methodName, duccId, messages.fetchLabel("unsuitable state")+serviceState);
							break;
						default:
							break;
						}
						break;
					default:
						break;
					}
				}
			}
			if(changes > 0) {
				OrchestratorCheckpoint.getInstance().saveState();
			}
		}
		ts.ended();	
		logger.trace(methodName, null, messages.fetch("exit"));
	}
	
	private int processPurger(DuccWorkJob job,Map<DuccId, IResource> map) {
		String methodName = "processPurger";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		if(job != null) {
			if(map != null) {
				Iterator<DuccId> iterator = map.keySet().iterator();
				while(iterator.hasNext()) {
					DuccId duccId = iterator.next();
					IResource resource = map.get(duccId);
					if(resource.isPurged()) {
						IDuccProcess process = job.getProcessMap().get(duccId);
						if(!process.isDefunct()) {
							String rState = process.getResourceState().toString();
							String pState = process.getProcessState().toString();
							logger.info(methodName, job.getDuccId(), duccId, "rState:"+rState+" "+"pState:"+pState);
							OrUtil.setResourceState(job, process, ResourceState.Deallocated);
							process.setProcessDeallocationType(ProcessDeallocationType.Purged);
							process.advanceProcessState(ProcessState.Stopped);
							changes++;
						}
					}
				}
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return changes;
	}
	
	private int processMapResourcesAdd(DuccWorkJob duccWorkJob, int memoryGbPerProcess, Map<DuccId,IResource> resourceMap) {
		String methodName = "processMapResourcesAdd";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		duccWorkJob.getSchedulingInfo().setMemorySizeAllocatedInBytes(memoryGbPerProcess*SizeBytes.GB);
		if(resourceMap == null) {
			logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("no map found"));
		}
		else {
			IDuccProcessMap processMap = duccWorkJob.getProcessMap();
			Iterator<DuccId> resourceMapIterator = resourceMap.keySet().iterator();
			while(resourceMapIterator.hasNext()) {
				DuccId duccId = resourceMapIterator.next();
				Node node = resourceMap.get(duccId).getNode();
				NodeIdentity nodeId = node.getNodeIdentity();
				if(!processMap.containsKey(duccId)) {
					ProcessType processType = null;
					switch(duccWorkJob.getServiceDeploymentType()) {
					case custom:
					case other:
						processType = ProcessType.Pop;
						break;
					case uima:
					case unspecified:
						processType = ProcessType.Job_Uima_AS_Process;
						break;
					}
					DuccProcess process = new DuccProcess(duccId, node, processType);
					long process_max_size_in_bytes = memoryGbPerProcess * SizeBytes.GB;
					CGroupManager.assign(duccWorkJob.getDuccId(), process, process_max_size_in_bytes);
					orchestratorCommonArea.getProcessAccounting().addProcess(duccId, duccWorkJob.getDuccId());
					processMap.addProcess(process);
					OrUtil.setResourceState(duccWorkJob, process, ResourceState.Allocated);
					logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("resource added")
												+" "+messages.fetchLabel("process")+duccId.getFriendly()
												+" "+messages.fetchLabel("unique")+duccId.getUnique()
												+" "+messages.fetchLabel("name")+nodeId.getName()
												+" "+messages.fetchLabel("ip")+nodeId.getIp());
					changes++;
					// check on usefulness of recent allocation
					switch(duccWorkJob.getJobState()) {
					// allocation unnecessary if job is already completed
					case Completing:
					case Completed:
						OrUtil.setResourceState(duccWorkJob, process, ResourceState.Deallocated);
						process.setProcessDeallocationType(ProcessDeallocationType.Voluntary);
						process.advanceProcessState(ProcessState.Stopped);
						logger.warn(methodName, duccWorkJob.getDuccId(), 
								messages.fetch("resource allocated for completed job")
								+" "+
								messages.fetchLabel("process")+duccId.getFriendly()
								);
						break;
					default:
						// allocation unnecessary if job has excess capacity
						if(isExcessCapacity(duccWorkJob)) {
							OrUtil.setResourceState(duccWorkJob, process, ResourceState.Deallocated);
							process.setProcessDeallocationType(ProcessDeallocationType.Voluntary);
							process.advanceProcessState(ProcessState.Stopped);
							logger.warn(methodName, duccWorkJob.getDuccId(), 
									messages.fetch("resource allocated for over capacity job")
									+" "+
									messages.fetchLabel("process")+duccId.getFriendly()
									);
						}
						break;
					}
				}
				else {
					logger.warn(methodName, duccWorkJob.getDuccId(), messages.fetch("resource exists")
						+" "+messages.fetchLabel("process")+duccId.getFriendly()
						+" "+messages.fetchLabel("unique")+duccId.getUnique()
						+" "+messages.fetchLabel("name")+nodeId.getName()
						+" "+messages.fetchLabel("ip")+nodeId.getIp());
				}
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return changes;
	}
	
	private int processMapResourcesDel(DuccWorkJob duccWorkJob, int memoryGbPerProcess, Map<DuccId,IResource> resourceMap) {
		String methodName = "processMapResourcesDel";
		logger.trace(methodName, duccWorkJob.getDuccId(), messages.fetch("enter"));
		int changes = 0;
		if(resourceMap == null) {
			logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("no map found"));
		}
		else {
			IDuccProcessMap processMap = duccWorkJob.getProcessMap();
			Iterator<DuccId> resourceMapIterator = resourceMap.keySet().iterator();
			logger.debug(methodName, duccWorkJob.getDuccId(), messages.fetchLabel("size")+processMap.size());
			while(resourceMapIterator.hasNext()) {
				DuccId duccId = resourceMapIterator.next();
				Node node = resourceMap.get(duccId).getNode();
				NodeIdentity nodeId = node.getNodeIdentity();
				logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("resource processing")
					+" "+messages.fetchLabel("process")+duccId.getFriendly()
					+" "+messages.fetchLabel("unique")+duccId.getUnique()
					+" "+messages.fetchLabel("name")+nodeId.getName()
					+" "+messages.fetchLabel("ip")+nodeId.getIp());
				if(processMap.containsKey(duccId)) {
					IDuccProcess process = processMap.get(duccId);
					switch(process.getResourceState()) {
					case Deallocated:
						break;
					default:
						OrUtil.setResourceState(duccWorkJob, process, ResourceState.Deallocated);
						process.setProcessDeallocationType(ProcessDeallocationType.Forced);
						logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("resource deallocated")
							+" "+messages.fetchLabel("process")+duccId.getFriendly()
							+" "+messages.fetchLabel("unique")+duccId.getUnique()
							+" "+messages.fetchLabel("name")+nodeId.getName()
							+" "+messages.fetchLabel("ip")+nodeId.getIp());
						break;
					}
					/*
					if(process.isDefunct()) {
						orchestratorCommonArea.getProcessAccounting().removeProcess(duccId);
						processMap.removeProcess(duccId);
						logger.info(methodName, duccId, messages.fetch("remove resource")+" "+messages.fetchLabel("name")+nodeId.getName()+" "+messages.fetchLabel("ip")+nodeId.getIp());
						changes++;
					}
					*/
				}
				else {
					logger.info(methodName, duccWorkJob.getDuccId(), messages.fetch("resource not found")
						+" "+messages.fetchLabel("process")+duccId.getFriendly()
						+" "+messages.fetchLabel("unique")+duccId.getUnique()
						+" "+messages.fetchLabel("name")+nodeId.getName()
						+" "+messages.fetchLabel("ip")+nodeId.getIp());
				}
			}
		}
		logger.trace(methodName, duccWorkJob.getDuccId(), messages.fetch("exit"));
		return changes;
	}

	private int reservationMapResourcesAdd(DuccWorkReservation duccWorkReservation,int memoryGbPerProcess,Map<DuccId,IResource> resourceMap) {
		String methodName = "reservationMapResourcesAdd";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		duccWorkReservation.getSchedulingInfo().setMemorySizeAllocatedInBytes(memoryGbPerProcess*SizeBytes.GB);
		IDuccReservationMap reservationMap = duccWorkReservation.getReservationMap();
		if(resourceMap != null) {
			Iterator<DuccId> resourceMapIterator = resourceMap.keySet().iterator();
			while(resourceMapIterator.hasNext()) {
				DuccId duccId = resourceMapIterator.next();
				IResource resource = resourceMap.get(duccId);
				Node node = resource.getNode();
				NodeIdentity nodeId = node.getNodeIdentity();
				if(!reservationMap.containsKey(duccId)) {
					long bytes = memoryGbPerProcess * SizeBytes.GB;
					DuccReservation reservation = new DuccReservation(duccId, node, bytes);
					reservationMap.addReservation(reservation);
					logger.info(methodName, duccId, messages.fetch("add resource")+" "+messages.fetchLabel("name")+nodeId.getName()+" "+messages.fetchLabel("ip")+nodeId.getIp());
					changes++;
				}
				else {
					logger.debug(methodName, duccId, messages.fetch("duplicate resource")+" "+messages.fetchLabel("name")+nodeId.getName()+" "+messages.fetchLabel("ip")+nodeId.getIp());
				}
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return changes;
	}
	
	private int reservationMapResourcesDel(DuccWorkReservation duccWorkReservation,int memoryGbPerProcess,Map<DuccId,IResource> resourceMap) {
		String methodName = "processMapResourcesDel";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		IDuccReservationMap reservationMap = duccWorkReservation.getReservationMap();
		if(resourceMap != null) {
			Iterator<DuccId> resourceMapIterator = resourceMap.keySet().iterator();
			while(resourceMapIterator.hasNext()) {
				DuccId duccId = resourceMapIterator.next();
				Node node = resourceMap.get(duccId).getNode();
				NodeIdentity nodeId = node.getNodeIdentity();
				if(reservationMap.containsKey(duccId)) {
					reservationMap.removeReservation(duccId);
					logger.info(methodName, duccId, messages.fetch("remove resource")+" "+messages.fetchLabel("name")+nodeId.getName()+" "+messages.fetchLabel("ip")+nodeId.getIp());
					changes++;
				}
				else {
					logger.debug(methodName, duccId, messages.fetch("not found resource")+" "+messages.fetchLabel("name")+nodeId.getName()+" "+messages.fetchLabel("ip")+nodeId.getIp());
				}
			}
		}
		logger.trace(methodName, null, messages.fetch("exit"));
		return changes;
	}
	
	/**
	 * SM reconciliation
	 */
	private String getServiceDependencyMessages(ServiceDependency sd) {
		String retVal = null;
		Map<String, String> messages = sd.getMessages();
		if(messages != null) {
			StringBuffer sb = new StringBuffer();
			for(String endpoint : messages.keySet()) {
				sb.append(endpoint);
				sb.append(":");
				sb.append(messages.get(endpoint));
				sb.append(";");
			}
			retVal = sb.toString();
		}
		return retVal;
	}
	
	public void reconcileState(ServiceMap serviceMap) {
		String methodName = "reconcileState (SM)";
		logger.trace(methodName, null, messages.fetch("enter"));
		int changes = 0;
		Iterator<DuccId> serviceMapIterator = serviceMap.keySet().iterator();
		TrackSync ts = TrackSync.await(workMap, this.getClass(), methodName);
		synchronized(workMap) {
			ts.using();
			while(serviceMapIterator.hasNext()) {
				DuccId duccId = serviceMapIterator.next();
				ServiceDependency services = serviceMap.get(duccId);
				DuccWorkJob duccWorkJob = (DuccWorkJob) WorkMapHelper.findDuccWork(workMap, duccId, this, methodName);
				if(duccWorkJob != null) {
					JobState jobState = duccWorkJob.getJobState();
					ServiceState serviceState = services.getState();
					switch(jobState) {
					case Received:
						logger.warn(methodName, duccId, messages.fetchLabel("unexpected job state")+jobState);
						break;
					case WaitingForDriver:
						logger.debug(methodName, duccId, messages.fetchLabel("pending job state")+jobState);
						break;
					case WaitingForServices:
						switch(serviceState) {
                        case Pending:                   // UIMA-4223
						case Waiting:
                        case Starting:
						case Initializing:
							break;
						case Available:
							stateJobAccounting.stateChange(duccWorkJob, JobState.WaitingForResources);
							changes++;
							logger.info(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
							break;
						case NotAvailable:
						case Stopped:
                        case Stopping:
							stateJobAccounting.stateChange(duccWorkJob, JobState.Completing);
							duccWorkJob.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
							String sdm = getServiceDependencyMessages(services);  
							IRationale rationale = new Rationale();
							if(sdm != null) {
								rationale = new Rationale("service manager reported "+sdm);
							}
							stateJobAccounting.complete(duccWorkJob, JobCompletionType.ServicesUnavailable, rationale);
							changes++;
							logger.info(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
							break;
						case Undefined:
							logger.warn(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
							break;
						}
						break;
					case WaitingForResources:
						logger.debug(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
						break;
					case Assigned:
					case Initializing:
					case Running:
						logger.debug(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
						break;
					case Completed:
						logger.debug(methodName, duccId, messages.fetchLabel("job state")+jobState+" "+messages.fetchLabel("services state")+serviceState);
						break;
					case Undefined:
						logger.warn(methodName, duccId, messages.fetchLabel("unexpected job state")+jobState);
						break;
					default:
						break;
					}
				}
				else {
					logger.debug(methodName, duccId, messages.fetch("job not found"));
				}
			}
			if(changes > 0) {
				OrchestratorCheckpoint.getInstance().saveState();
			}
		}
		ts.ended();	
		logger.trace(methodName, null, messages.fetch("exit"));
	}

	/**
	 * Node Inventory reconciliation
	 */
	public void reconcileState(HashMap<DuccId, IDuccProcess> inventoryProcessMap) {
		String methodName = "reconcileState (Node Inventory)";
		logger.trace(methodName, null, messages.fetch("enter"));
		Iterator<DuccId> iterator = inventoryProcessMap.keySet().iterator();
		TrackSync ts = TrackSync.await(workMap, this.getClass(), methodName);
		synchronized(workMap) {
			ts.using();
			while(iterator.hasNext()) {
				DuccId processId = iterator.next();
				IDuccProcess inventoryProcess = inventoryProcessMap.get(processId);
				List<IUimaPipelineAEComponent> upcList = inventoryProcess.getUimaPipelineComponents();
				if(upcList != null) {
					Iterator<IUimaPipelineAEComponent> upcIterator = upcList.iterator();
					while(upcIterator.hasNext()) {
						IUimaPipelineAEComponent upc = upcIterator.next();
						logger.trace(methodName, null, processId, "pipelineInfo: "+inventoryProcess.getNodeIdentity()+" "+inventoryProcess.getPID()+" "+upc.getAeName()+" "+upc.getAeState()+" "+upc.getInitializationTime());
					}
				}
				ProcessType processType = inventoryProcess.getProcessType();
				if(processType != null) {
					DuccId jobId = OrchestratorCommonArea.getInstance().getProcessAccounting().getJobId(processId);
					if(jobId != null) {
						IDuccWork duccWork = WorkMapHelper.findDuccWork(workMap, jobId, this, methodName);
						if(duccWork != null) {
							DuccType jobType = duccWork.getDuccType();
							switch(jobType) {
							case Job:
								DuccWorkJob job = (DuccWorkJob) duccWork;
								switch(processType) {
								case Pop:
									OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(inventoryProcess);
									switch(inventoryProcess.getProcessState()) {
									case LaunchFailed:
									case Failed:
										if(inventoryProcess.getDuccId().getFriendly() == 0) {
											jobTerminate(job, JobCompletionType.DriverProcessFailed, new Rationale(inventoryProcess.getReasonForStoppingProcess()), inventoryProcess.getProcessDeallocationType());
										}
										else {
											jobTerminate(job, JobCompletionType.ProcessFailure, new Rationale(inventoryProcess.getReasonForStoppingProcess()), inventoryProcess.getProcessDeallocationType());
										}
										break;
									default:
										if(inventoryProcess.isComplete()) {
											OrchestratorCommonArea.getInstance().getProcessAccounting().deallocate(job,ProcessDeallocationType.Stopped);
											IRationale rationale = new Rationale("state manager reported as normal completion");
											int errors = job.getSchedulingInfo().getIntWorkItemsError();
											int done = job.getSchedulingInfo().getIntWorkItemsCompleted();
											if(errors > 0) {
												setCompletionIfNotAlreadySet(job, JobCompletionType.Error, new Rationale("state manager detected error work items="+errors));
											}
											else if(done == 0) {
												setCompletionIfNotAlreadySet(job, JobCompletionType.EndOfJob, new Rationale("state manager detected no work items processed"));
											}
											// <UIMA-3337>
											else {
												setCompletionIfNotAlreadySet(job, JobCompletionType.EndOfJob, rationale);
											}
											// </UIMA-3337>
											completeJob(job, rationale);
										}
										break;
									}
									break;
								case Service:
									logger.warn(methodName, jobId, processId, "unexpected process type: "+processType);
									break;
								case Job_Uima_AS_Process:
									OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(inventoryProcess);
									try {
										if(!job.isInitialized()) {
											IDuccProcessMap map = job.getProcessMap();
											for(Entry<DuccId, IDuccProcess> entry : map.entrySet()) {
												IDuccProcess process = entry.getValue();
												StringBuffer sb = new StringBuffer();
												sb.append("pid:"+process.getPID()+" ");
												sb.append("state:"+process.getProcessState()+" ");
												sb.append("reason:"+process.getReasonForStoppingProcess()+" ");
												logger.trace(methodName, job.getDuccId(), sb.toString());
											}
											long initFailureCount = job.getProcessInitFailureCount();
											long startup_initialization_error_limit = DuccPropertiesResolver.get(DuccPropertiesResolver.ducc_jd_startup_initialization_error_limit, 1);
											if(initFailureCount >= startup_initialization_error_limit) {
												String reason = "process inititialization failure count["+initFailureCount+"] meets startup initialization error limit["+startup_initialization_error_limit+"]";
												logger.warn(methodName, job.getDuccId(), reason);
												JobCompletionType jobCompletionType = JobCompletionType.CanceledBySystem;
												Rationale rationale = new Rationale(reason);
												ProcessDeallocationType processDeallocationType = ProcessDeallocationType.JobCanceled;
												stateManager.jobTerminate(job, jobCompletionType, rationale, processDeallocationType);
											}
											else {
												String reason = "process failure count["+initFailureCount+"] does not exceed startup initialization error limit["+startup_initialization_error_limit+"]";
												logger.debug(methodName, job.getDuccId(), reason);
											}
										}
										else {
											logger.trace(methodName, job.getDuccId(), "job is initialized");
										}
									}
									catch(Exception e) {
										logger.error(methodName, jobId, e);
									}
									break;
								}
								// <UIMA-3923>
								advanceToCompleted(job);
								// </UIMA-3923>
								break;
							case Service:
								DuccWorkJob service = (DuccWorkJob) duccWork;
								switch(processType) {
								case Pop:
									OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(inventoryProcess);
									if(inventoryProcess.isComplete()) {
										OrchestratorCommonArea.getInstance().getProcessAccounting().deallocate(service,ProcessDeallocationType.Stopped);
									}
									if(!service.hasAliveProcess()) {
										completeManagedReservation(service, new Rationale("state manager reported no viable service process exists, type="+processType));
									}
									break;
								case Service:
									OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(inventoryProcess);
									if(inventoryProcess.isComplete()) {
										OrchestratorCommonArea.getInstance().getProcessAccounting().deallocate(service,ProcessDeallocationType.Stopped);
									}
									if(!service.hasAliveProcess()) {
										completeService(service, new Rationale("state manager reported no viable service process exists, type="+processType));
									}
									break;
								case Job_Uima_AS_Process:
									OrchestratorCommonArea.getInstance().getProcessAccounting().setStatus(inventoryProcess);
									if(inventoryProcess.isComplete()) {
										OrchestratorCommonArea.getInstance().getProcessAccounting().deallocate(service,ProcessDeallocationType.Stopped);
									}
									if(!service.hasAliveProcess()) {
										completeService(service, new Rationale("state manager reported no viable service process exists, type="+processType));
									}
									break;
								}
								// <UIMA-3923>
								advanceToCompleted(service);
								// </UIMA-3923>
								break;
							default:
								break;
							}
						}
						else {
							StringBuffer sb = new StringBuffer();
							sb.append("node:"+inventoryProcess.getNodeIdentity().getName());
							sb.append(" ");
							sb.append("PID:"+inventoryProcess.getPID());
							sb.append(" ");
							sb.append("type:"+inventoryProcess.getProcessType());
							logger.debug(methodName, jobId, sb);
						}
					}
					else {
						StringBuffer sb = new StringBuffer();
						sb.append("node:"+inventoryProcess.getNodeIdentity().getName());
						sb.append(" ");
						sb.append("PID:"+inventoryProcess.getPID());
						sb.append(" ");
						sb.append("type:"+inventoryProcess.getProcessType());
						logger.debug(methodName, jobId, sb);
					}
				}
				else {
					DuccId jobId = null;
					StringBuffer sb = new StringBuffer();
					sb.append("node:"+inventoryProcess.getNodeIdentity().getName());
					sb.append(" ");
					sb.append("PID:"+inventoryProcess.getPID());
					sb.append(" ");
					sb.append("type:"+inventoryProcess.getProcessType());
					logger.warn(methodName, jobId, sb);
				}
			}
		}
		ts.ended();
		logger.trace(methodName, null, messages.fetch("exit"));
	}
	
	private void advanceToCompleted(DuccWorkJob job) {
		switch(job.getJobState()) {
		case Completing:
			if(job.getProcessMap().getAliveProcessCount() == 0) {
				stateJobAccounting.stateChange(job, JobState.Completed);
			}
			break;
		default:
			break;
		}
	}
	
	private void advanceToCompleting(DuccWorkJob job) {
		switch(job.getJobState()) {
		case Completing:
		case Completed:
			break;
		default:
			if(job.getProcessMap().getAliveProcessCount() == 0) {
				stateJobAccounting.stateChange(job, JobState.Completing);
			}
		}
	}
	
	private void completeManagedReservation(DuccWorkJob service, IRationale rationale) {
		String location = "completeManagedReservation";
		DuccId jobid = null;
		try {
			jobid = service.getDuccId();
			Map<DuccId, IDuccProcess> map = service.getProcessMap().getMap();
			int size = map.size();
			if(size != 1) {
				logger.warn(location, jobid, "size: "+size);
				completeJob(service, rationale);
			}
			else {
				Iterator<DuccId> iterator = map.keySet().iterator();
				while(iterator.hasNext()) {
					DuccId key = iterator.next();
					IDuccProcess process = map.get(key);
					int code = process.getProcessExitCode();
					IRationale exitCode = new Rationale("code="+code);
					switch(service.getCompletionType()) {
					case Undefined:
						JobCompletionType completionType = JobCompletionType.ProgramExit;
						ProcessState processState = process.getProcessState();
						if(processState != null) {
							switch(processState) {
							case LaunchFailed:
								completionType = JobCompletionType.LaunchFailure;
								break;
							default:
								break;
							}
						}
						service.setCompletion(completionType, exitCode);
						service.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
						break;
					default:
						break;
					}
					advanceToCompleting(service);
					break;
				}
			}
		}
		catch(Exception e) {
			logger.error(location, jobid, e);
			completeJob(service, rationale);
		}
	}
	
	private void completeService(DuccWorkJob service, IRationale rationale) {
		String location = "completeService";
		DuccId jobid = service.getDuccId();
		if(service.getProcessFailureCount() > 0) {
			service.setCompletion(JobCompletionType.Warning, new Rationale("process failure(s) occurred"));
			logger.debug(location, jobid, service.getCompletionRationale().getText()+", "+"ProcessFailureCount="+service.getProcessFailureCount());
		}
		else if(service.getProcessInitFailureCount() > 0) {
			service.setCompletion(JobCompletionType.Warning, new Rationale("process initialization failure(s) occurred"));
			logger.debug(location, jobid, service.getCompletionRationale().getText()+", "+"ProcessInitFailureCount="+service.getProcessInitFailureCount());
		}
		else {
			setCompletionIfNotAlreadySet(service, JobCompletionType.EndOfJob, rationale);
			logger.debug(location, jobid, service.getCompletionRationale().getText()+", "+"no failures");
		}
		advanceToCompleting(service);
	}
	
	private void completeJob(DuccWorkJob job, IRationale rationale) {
		String location = "completeJob";
		DuccId jobid = null;
		switch(job.getCompletionType()) {
		case Undefined:
			job.setCompletion(JobCompletionType.Undefined, rationale);
			job.getStandardInfo().setDateOfCompletion(TimeStamp.getCurrentMillis());
			break;
		case EndOfJob:
			if(job.getProcessFailureCount() > 0) {
				job.setCompletion(JobCompletionType.Warning, new Rationale("all work items completed, but job process failure(s) occurred"));
			}
			else if(job.getProcessInitFailureCount() > 0) {
				job.setCompletion(JobCompletionType.Warning, new Rationale("all work items completed, but job process initialization failure(s) occurred"));
			}
			else {
				try {
					if(Integer.parseInt(job.getSchedulingInfo().getWorkItemsError()) > 0) {
						job.setCompletion(JobCompletionType.Error, rationale);
					}
				}
				catch(Exception e) {
					logger.error(location, jobid, e);
				}
			}
			break;
		default:
			break;
		}
		advanceToCompleting(job);
	}
	
	public void jobTerminate(IDuccWorkJob job, JobCompletionType jobCompletionType, IRationale rationale, ProcessDeallocationType processDeallocationType) {
		if(!job.isFinished()) {
			stateJobAccounting.stateChange(job, JobState.Completing);
			stateJobAccounting.complete(job, jobCompletionType, rationale);
			OrchestratorCommonArea.getInstance().getProcessAccounting().deallocate(job,processDeallocationType);
		}
	}
	
}