package gov.va.med.esr.common.batchprocess;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Iterator;
import java.util.List;


import org.springframework.util.StringUtils;

import gov.va.med.esr.common.infra.ImpreciseDate;
import gov.va.med.esr.common.model.lookup.Relationship;
import gov.va.med.esr.common.model.lookup.SSAVerificationStatus;
import gov.va.med.esr.common.model.lookup.SSNType;
import gov.va.med.esr.common.model.person.SSNVerification;
import gov.va.med.esr.common.model.person.SSNVerificationDetail;

import gov.va.med.esr.common.persistent.person.SSNVerificationDetailDAO;
import gov.va.med.esr.service.DemographicService;
import gov.va.med.esr.service.LookupService;

import gov.va.med.esr.service.UniqueIdentifierGenerator;
import gov.va.med.fw.batchprocess.AbstractDataQueryIncrementalProcess;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;

import gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext;
import gov.va.med.fw.batchprocess.ProcessStatistics;
import gov.va.med.fw.io.writer.FormattedFileWriter;
import gov.va.med.fw.persistent.DAOException;
import gov.va.med.fw.service.ServiceException;
import gov.va.med.fw.util.ThreadPool;

/**
 * Contains logic to process the Request Person SSN Verification use case.
 * @author Rajiv Patnaik Created on Mar 7, 2006
 * @version 1.0
 *
 * Copyright © 2006 VHA. All rights reserved
 */
public class RequestSSNVerificationProcess extends
        AbstractDataQueryIncrementalProcess
{
    //BEGIN - thread related properties
    private static final int DEFAULT_THREAD_POOL_SIZE = 10;
	private static final int DEFAULT_THROTTLE_TASKCOUNT_THREASHOLD = 100;
	private static final int DEFAULT_SPAWN_RETRY_PERIOD = 3000;
	private static final int DEFAULT_EXCEPTION_UPDATE_INTERVAL = 20;
	public static final String CONTEXT_THREAD_CREATOR = "threadCreator";
	public static final String CONTEXT_TASK_COUNT = "taskCount";
	public static final String CONTEXT_THREAD_POOL = "threadPool";
	
	private int threadPoolSize = DEFAULT_THREAD_POOL_SIZE;
	private int throttleTaskCountThreshold = DEFAULT_THROTTLE_TASKCOUNT_THREASHOLD;
	private int spawnRetryPeriod = DEFAULT_SPAWN_RETRY_PERIOD;
	private String spawnedTaskId; 
	//END thread related properties
	
	
    /**
     * Key to be token replaced for dynamically created parameter holders (?,?)
     * needed to execute an update on SSN table.
     */
    private static String PARAMETER_HOLDER = "PARAMETER_HOLDER";
    private static SSAVerificationStatus SSN_VERIFICATION_IN_PROCESS = null;
    
    //max number of records per file
    private static int MAX_RECORD_PER_FILE = 250000;
    
	//spring injection from batch_processes.xml
    private String updateQueryString;
    private LookupService lookupService;
    private DemographicService demographicService;
    private FormattedFileWriter outputFileWriter;
    private UniqueIdentifierGenerator generator;
    private SSNVerificationDetailDAO ssnVerificationDetailDAO;
    
    
    //TESTING ONLY. PLEASE REMOVE!!!
    public void setMasRecordPerFile(int max)
    {
    	MAX_RECORD_PER_FILE = max;
    }
    /**
     * @return Returns the demographicService.
     */
    public DemographicService getDemographicService()
    {
        return demographicService;
    }

    /**
     * @param demographicService
     *            The demographicService to set.
     */
    public void setDemographicService(DemographicService demographicService)
    {
        this.demographicService = demographicService;
    }

    /**
     * @return Returns the lookupService.
     */
    public LookupService getLookupService()
    {
        return lookupService;
    }

    /**
     * @param lookupService
     *            The lookupService to set.
     */
    public void setLookupService(LookupService lookupService)
    {
        this.lookupService = lookupService;
    }

	public SSNVerificationDetailDAO getSsnVerificationDetailDAO() {
		return ssnVerificationDetailDAO;
	}

	public void setSsnVerificationDetailDAO(
			SSNVerificationDetailDAO ssnVerificationDetailDAO) {
		this.ssnVerificationDetailDAO = ssnVerificationDetailDAO;
	}

    /**
     * @return Returns the outputFileWriter.
     */
    public FormattedFileWriter getOutputFileWriter()
    {
        return outputFileWriter;
    }

    /**
     * @param outputFileWriter
     *            The outputFileWriter to set.
     */
    public void setOutputFileWriter(FormattedFileWriter outputFileWriter)
    {
        this.outputFileWriter = outputFileWriter;
    }

    public UniqueIdentifierGenerator getGenerator() {
		return generator;
	}

	public void setGenerator(UniqueIdentifierGenerator generator) {
		this.generator = generator;
	}

	/**
     * @return Returns the updateQueryString.
     */
    public String getUpdateQueryString()
    {
        return updateQueryString;
    }

    private String generateControlIdentifier() throws ServiceException {
        return this.generator.generate().toString();
     }
    
    /**
     * @param updateQueryString
     *            The updateQueryString to set.
     */
    public void setUpdateQueryString(String updateQueryString)
    {
        this.updateQueryString = updateQueryString;
    }

    protected DataProcessExecutionContext createDataProcessExecutionContext()
    {
        DataQueryProcessExecutionContext context = new SSNVerificationProcessExecutionContext();
        context.setProcessStatistics(createProcessStatistics());
        return context;
    }

	protected ProcessStatistics createProcessStatistics() {
		return new RequestSSNVerificationStatistics();
	}

	/********* BEGIN: thread related methods *****************/
	
	public String getSpawnedTaskId() {
		return spawnedTaskId;
	}

	public void setSpawnedTaskId(String spawnedTaskId) {
		this.spawnedTaskId = spawnedTaskId;
	}
	/**
	 * @return Returns the threadPoolSize.
	 */
	public int getThreadPoolSize() {
		return threadPoolSize;
	}

	/**
	 * @param threadPoolSize The threadPoolSize to set.
	 */
	public void setThreadPoolSize(int threadPoolSize) {
		this.threadPoolSize = threadPoolSize;
	}
	
	private boolean isThreaded() {
		return gov.va.med.fw.util.StringUtils.isNotBlank(spawnedTaskId);
	}

	private void cleanThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = getThreadPool(context);
		threadPool.stop();
		context.getContextData().put(CONTEXT_THREAD_POOL, null);
	}

	protected void handleDataProcessCompleted(
			DataProcessExecutionContext context) {
		if (isThreaded()) {
			try {
				if (getTaskCount(context) != 0) {
					synchronized (this) {
						boolean stillProcessing = true;
						while (stillProcessing) {
					        if ( logger.isDebugEnabled())
					        {
					        	logger.debug(new Date() + " RequestSSNVerificationProcess.handleDataProcessCompleted() : taskCount before wait: "  + getTaskCount(context));
					        }
							wait();
					        if ( logger.isDebugEnabled())
					        {
					        	logger.debug(new Date() + " RequestSSNVerificationProcess.handleDataProcessCompleted() : taskCount after wait: "  + getTaskCount(context));
					        }

							if (getTaskCount(context) == 0)
								stillProcessing = false;
						}
					}
				}
			} catch (InterruptedException e) {
				throwIllegalStateException(
						"RequestSSNVerificationProcess was interrupted while it was waiting for "
								+ "its spawned threads to complete", e);
			} finally {
				getThreadPool(context).stop();
				cleanThreadPool(context);
			}
		}
		
		
		List vetFileOutputData = ((SSNVerificationProcessExecutionContext) context).getVetFileOutputData();
        RequestSSNVerificationStatistics stats = (RequestSSNVerificationStatistics) context.getProcessStatistics();

        if (vetFileOutputData != null && vetFileOutputData.size() > 0)
        {
	        synchronized (this)
	        {
	        	outputFileWriter.appendData(vetFileOutputData);
	        }
	        
			// Set the successful records stats
			stats.setNumberOfSuccessfulRecords(stats.getNumberOfSuccessfulRecords()
					+ vetFileOutputData.size());
			stats.setNumberOfSuccessfulVetRecords(stats
					.getNumberOfSuccessfulVetRecords()
					+ vetFileOutputData.size());
        }
        
		stats.setVetProcessingEndDate(new Date());		// update the batch stats

		// update stats
		this.updateJobResult(context);

		super.handleDataProcessCompleted(context);
	}

	public int getThrottleTaskCountThreshold() {
		return throttleTaskCountThreshold;
	}

	public void setThrottleTaskCountThreshold(int throttleTaskCountThreshold) {
		this.throttleTaskCountThreshold = throttleTaskCountThreshold;
	}

	public int getSpawnRetryPeriod() {
		return spawnRetryPeriod;
	}

	public void setSpawnRetryPeriod(int spawnRetryPeriod) {
		this.spawnRetryPeriod = spawnRetryPeriod;
	}

	private ThreadPool getThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = (ThreadPool) context.getContextData().get(
				CONTEXT_THREAD_POOL);
		if (threadPool == null) {
			threadPool = new ThreadPool(
					"RequestSSNVerificationProcessSpawnedThreadTask", threadPoolSize);
			context.getContextData().put(CONTEXT_THREAD_POOL, threadPool);
		}
		return threadPool;
	}

	private void initThreadCreatorAndIncrementTaskCount(
			DataQueryProcessExecutionContext context) {
		if (!context.getContextData().containsKey(CONTEXT_THREAD_CREATOR))
			context.getContextData().put(CONTEXT_THREAD_CREATOR, this);

		adjustTaskCount(context, 1);
	}

	static int getTaskCount(DataProcessExecutionContext context) {
		Integer count = (Integer) context.getContextData().get(
				CONTEXT_TASK_COUNT);
		return count != null ? count.intValue() : 0;
	}

	static void adjustTaskCount(DataQueryProcessExecutionContext context,
			int adjustment) {
		synchronized (context) {
			context.getContextData().put(CONTEXT_TASK_COUNT,
					new Integer(getTaskCount(context) + adjustment));
		}
	}
    /******* END: thread related methods *****************/


    /**
     * This gets data from 2 sources. 
     * 1. retrieve all veteran records from SSN Verification Queue (ssn_verification_detail)
     * 2. retrieve all spouse and dependent records from ESR (ssn) with record status of "New". 
     * Convert data from both sources to type SSASSNVerificationData.
     *
     * The spouse and dependents data will be processed incrementally with the
     * existing framework support as they are retrieved friom ESR. The vet data
     * will be processed in one shot.
     */
    protected List doAcquireData(DataQueryProcessExecutionContext context)
            throws Exception
    {
        
        SSNVerificationProcessExecutionContext ssnContext = (SSNVerificationProcessExecutionContext) context;
        
        //reach max file size, stop
        if (ssnContext.isReachMaxSize())
        	return null;

        
        //DNS   doank - ESR_CodeCR8407
        //allow execution argument to execute only spouse and dependents verifications
        Object arg = ssnContext.getExecutionArguments();
        boolean onlySpouseAndDependentOption = false;
        if( arg != null && arg instanceof String && arg.toString().equals("-sd"))
        	onlySpouseAndDependentOption = true;
        
        RequestSSNVerificationStatistics stats = (RequestSSNVerificationStatistics) context.getProcessStatistics();
        List list = null;
        if (!onlySpouseAndDependentOption && !ssnContext.isVeteranDataAcquired())
        {
        	//Make a one time call to get the veteran records from SSN Verification Queue
        	list = ssnVerificationDetailDAO.getSSNVerificationListToProcess();
        	
            int size = (list == null) ? 0: list.size();
            if (size > this.MAX_RECORD_PER_FILE) 
            {
            	size = MAX_RECORD_PER_FILE;
            	list = list.subList(0, MAX_RECORD_PER_FILE-1);
            	ssnContext.setReachMaxSize(true);
            }
            
            stats.setNumberOfTotalVetRecords(size);
            stats.setNumberOfTotalRecords(stats.getNumberOfTotalRecords()+ size);
           
            //Set the flag to true so that veteran data will not be acquired
            // in the next incremental call when spouse and dependents data is retieved.
            ssnContext.setVeteranDataAcquired(true);
            
            // in this run we are not supplying the S&D Data 
            ssnContext.setHasNoSpouseDependents(true);
            

            //If spouseAndDependentsData == 0, and the veteran data is not 0, we
            //want the process to continue. So set the acquireddata to the
            //veteran data, so that the parent class can call the processData method
            
            // for veteran data we get the data from the context so do not set the value here.
            
        } 
        else
        {
        	// only to process spouse and dependents option
        	if(onlySpouseAndDependentOption)
        		ssnContext.setVeteranDataProcessed(true);
            
        	// After the first iteration where we processed all the veterans data,
            // we are going to handle the S&D data using the serverside cursor.
            ssnContext.setHasNoSpouseDependents(false);
            Object[] params = new Object[3];
            params[0] = Integer.valueOf(SSAVerificationStatus.NEW_RECORD.getCode());
            //CCR11666 include RESEND_TO_SSA
            params[1] = Integer.valueOf(SSAVerificationStatus.RESEND_TO_SSA.getCode());
            params[2] = SSNType.CODE_ACTIVE.getCode();

            context.getCurrentDataQuery().getQuery().setParamValues(params);
            
            list = super.doAcquireData(context);
            
        }
        
        return list;
    }

    /**
     * 1. Write data to output file 2. Update the SSN statuses for veteran in
     * PSIM. 3. Update the SSN statuses for spouse and dependents in the ESR SSN
     * table 4. Insert summary info into SSNVerification and
     * SSNVerificationDetail tables
     *
     * The data will be processed incrementally. So each chunk(increment) of
     * data will be in a separate transaction.
     *
     * No transaction for this method.
     *
     */
    protected void processData(DataQueryProcessExecutionContext context,
            List acquiredData)
    {
        try
        {
            if (acquiredData != null)
            {
            	if (SSN_VERIFICATION_IN_PROCESS == null)
            		SSN_VERIFICATION_IN_PROCESS = getLookupService().getSSAVerificationStatusByCode(
                        SSAVerificationStatus.IN_PROCESS.getCode());
    			
                SSNVerificationProcessExecutionContext ssnContext =
                    (SSNVerificationProcessExecutionContext) context;


                //Get the SSNVerification object if it already exists
                //if not existing, create a new one, save it to the database, store it in the context
                SSNVerification ssnVerification = getSSNVerification(ssnContext);

                //Process veterans if it has not been processed yet
    			if (!ssnContext.isVeteranDataProcessed()) {
    				ssnVerification = processVeterans(ssnContext, ssnVerification, acquiredData);

    				// set VeteranDataProcessed true in the context so that we do
    				// not process them from the 2nd iteration onwards
    				ssnContext.setVeteranDataProcessed(true);
    			}

				// if the process is interrupted let us stop here. We only check this flag in 
    			//the context as we did a real time check at the end of the processVeterans method.
				if (ssnContext.isInterrupted()) {
					return;
				}

                //Process Spouse and dependents data 
                if ( ssnContext.hasNoSpouseDependents() == false )
                {   
                	//update the total number and stop at the max size
                	int size = (acquiredData == null) ? 0: acquiredData.size();
	                if (ssnContext.getProcessStatistics().getNumberOfTotalRecords() + size > this.MAX_RECORD_PER_FILE) 
	                {
	                	acquiredData = acquiredData.subList(0, MAX_RECORD_PER_FILE - ssnContext.getProcessStatistics().getNumberOfTotalRecords() - 1);
	                	ssnContext.getProcessStatistics().setNumberOfTotalRecords(MAX_RECORD_PER_FILE);
	                	ssnContext.setReachMaxSize(true);
	                } else
	                {
	                	ssnContext.getProcessStatistics().setNumberOfTotalRecords(ssnContext.getProcessStatistics().getNumberOfTotalRecords()+ size);
	                }   
	                
	                List spouseAndDependentsData =  convertToOutputData(acquiredData);
                    
                    ssnVerification = processSpouseAndDependents(ssnContext,
                            ssnVerification, spouseAndDependentsData);
                }

            }
        } catch (Exception e)
        {
            context.getProcessStatistics().setNumberOfErrorRecords(
                    context.getProcessStatistics()
                            .getNumberOfErrorRecords()
                            + acquiredData.size());

            String errorMessage = "Error while executing Request SSN Verification batch process :"
                    + " Reason: " + e.getMessage();

            context.getExceptionData().add(errorMessage);

            if (logger.isWarnEnabled())
                logger.warn(errorMessage, e);
        }
    }

	private SSNVerification getSSNVerification(SSNVerificationProcessExecutionContext ssnContext) throws DAOException {
		if (ssnContext.getSsnVerification() != null)
		{
			//existing ssnVerification, return it
		    return ssnContext.getSsnVerification();
		} 
	
		//new, save it so the PK is generated 
		SSNVerification ssnVerification = new SSNVerification();
	    ssnVerification.setCountSentToSsa(new Integer(0));
	    getDao().saveObject(ssnVerification);
	    ssnContext.setSsnVerification(ssnVerification);
	
		return ssnVerification;
	}

    /**
     * Update SSN table
     * Insert in SSN Verification table
     *
     * @param context
     * @param ssnVerification
     * @param spouseAndDependentsData
     * @return SSNVerification
     */
    private SSNVerification processSpouseAndDependents(
            SSNVerificationProcessExecutionContext context,
            SSNVerification ssnVerification, List spouseAndDependentsData)
    {
        
        
        RequestSSNVerificationStatistics stats = (RequestSSNVerificationStatistics) context.getProcessStatistics();
        try
        {
            String updatedQueryString = null;
            Object[] spouseAndDependentsSSNIds = null;

            if (spouseAndDependentsData != null && !spouseAndDependentsData.isEmpty())
            {

                //Create comma separated parameter holders (?,?...) to be
                // substituted in query
                String parameterHolder = createCommaSeparatedParameterHolders(spouseAndDependentsData);
                updatedQueryString = StringUtils.replace(
                        getUpdateQueryString(), PARAMETER_HOLDER,
                        parameterHolder);
                spouseAndDependentsSSNIds = getParameterData(
                        spouseAndDependentsData, SSN_VERIFICATION_IN_PROCESS);

                List newVerificationDetails = createSummaryStatisticsData(ssnVerification, spouseAndDependentsData);
                //Update SSN table with Verification Status = In Progress
                //Insert into the SSNVerification and Verification detail table
                ssnVerification = demographicService
                        .updateRequestSSNVerificationData(ssnVerification,newVerificationDetails,
                                updatedQueryString, spouseAndDependentsSSNIds);
                
            }
          

            //Get the primary key for SSNVerification and store it in the
            //execution context. This will enable an update for the next
            // batch of data
            if(context.getSsnVerification() == null)
                context.setSsnVerification(ssnVerification);


            //Write to output file
            if(spouseAndDependentsData != null && !spouseAndDependentsData.isEmpty())
                outputFileWriter.appendData(spouseAndDependentsData);

            //Set the successful records stats
            context.getProcessStatistics().setNumberOfSuccessfulRecords(
                    context.getProcessStatistics().getNumberOfSuccessfulRecords()
                    + spouseAndDependentsData.size());
            
            stats.setNumberOfSuccessfulSDRecords(
                    stats.getNumberOfSuccessfulSDRecords() + spouseAndDependentsData.size() );

            
            
            
            //Set the current record to 0 in the last step. This is to simulate idempotent behaviour. Say the total number of records in the SSN table
            //with the status "New Record" is 245. The first incremental query should return 100, the  2nd one 100, the third one 45. The query works by looking at
            // the current  record and then fetches the next 100 after that. This this  process updates the SSN table after processing each chunk and updates the SSN status
            // to "In Progress". So after the first chunk of records get processed, we have 145
            // records in the database with a status of "New Record". But since the current record
            // is now set at 100, the query will return the next chunk of records having a rownum > 100. That means we will
            // get the last 45 records and 100 records in between will be ignored. To get around it, set the current
            // record count to 0 at the end of process. If any chunk fails during processing, then this part of the
            // code will not be reached(as this is the last step). So current record will stay as-is and the next chunk of
            // record will be retrieved and this one will not be re-retrieved again.
            
            //DONOT HAVE TO reset the counter as implementing the serverside cusrsors from now so commenting the below
            
            //if (!context.hasNoSpouseDependents())
            //    context.setCurrentRecord(0);


        } catch (Exception e)
        {
            context.getProcessStatistics().setNumberOfErrorRecords(
                    context.getProcessStatistics()
                            .getNumberOfErrorRecords()
                            + (spouseAndDependentsData == null ? 0 :spouseAndDependentsData.size()));

            stats.setNumberOfErrorSDRecords(
                    stats.getNumberOfErrorSDRecords() + (spouseAndDependentsData == null ? 0 :spouseAndDependentsData.size()) );

            String errorMessage = "Error while executing Request SSN Verification batch process for spouse and dependents :"
                    + " Reason: " + e.getMessage();

            context.getExceptionData().add(errorMessage);

            if (logger.isWarnEnabled())
                logger.warn(errorMessage, e);
        }
        return ssnVerification;
    }

    /**
     * @param spouseAndDependentsData
     * @return the comma separated parameter holders.
     */
    private String createCommaSeparatedParameterHolders(
            List spouseAndDependentsData)
    {
        StringBuffer parameterHolder = new StringBuffer();
        for (int i = 0; i < spouseAndDependentsData.size(); i++)
        {
            parameterHolder.append("?");
            if (i != spouseAndDependentsData.size() - 1)
            {
                parameterHolder.append(",");
            }
        }
        return parameterHolder.toString();
    }

    /**
     * Creates the data for the update statement. Substitutes the first
     * parameter with the SSAVerificationStatus ans the rest with the Spouse and
     * Dependents SSN Id.
     *
     * @param spouseAndDependentsData
     * @param status
     * @return Object[]
     */
    private Object[] getParameterData(List spouseAndDependentsData,
            SSAVerificationStatus status)
    {
    	   Object[] parameterData = new Object[spouseAndDependentsData.size() + 1];
           parameterData[0] = status;
           int i = 1;
           for (Iterator iter = spouseAndDependentsData.iterator(); iter.hasNext();)
           {
               SSASSNVerificationData data = (SSASSNVerificationData) iter.next();
               parameterData[i] = new BigDecimal(data.getSsnId());
               i++;
           }
           return parameterData;
  }

    /**
     * Divide the total VPIDs received into smaller chunks and process a chunk at a time.
     * Notify PSIM to update SSN status
     * Insert into SSNVerification table
     *
     * @param context the ssn verification process execution context.
     * @param ssnVerification the ssn verification.
     *
     * @return the ssn verification.
     */
    private SSNVerification processVeterans(
            SSNVerificationProcessExecutionContext context, SSNVerification ssnVerification, List acquiredData)
    {
		RequestSSNVerificationStatistics stats = (RequestSSNVerificationStatistics) context
		.getProcessStatistics();
		
        if(acquiredData == null || acquiredData.isEmpty())
            return ssnVerification;

        stats.setVetProcessingStartDate(new Date());
        
        if ( logger.isDebugEnabled())
        {
        	logger.debug(" RequestSSNVerificationProcess.processVeterans() : Total number of Veterans Received: "  + acquiredData.size());
        }
        
		// update stats before multi-thread starts
		this.updateJobResult(context);
		
        List vetFileOutputData = new ArrayList();
         
        SSNVerificationDetail detail = null;
        		
        //loop through each veteran record, handle each record with a thread task
        //the veteran file output data is stored in List vetFileOutputData
        for (int i = 0; i < acquiredData.size() && !isInterrupted(context); i++)
        {
        	detail = (SSNVerificationDetail)acquiredData.get(i);
        	detail.setSsnVerification(ssnVerification); //make sure the FK to SSNVerification is assigned
        	
        	Object[] data = {detail, stats, vetFileOutputData};
        	
           	//it's veteran record, spawn a thread task to produce a veteran record
        	spawnThread(context, data);
        	        	
            if ( logger.isDebugEnabled())
            {
            	logger.debug(" RequestSSNVerificationProcess.processVeterans() i= "  + i);
            }
        }
        
        //store the veteran file output data in context and write it out in handleDataProcessCompleted()
        context.setVetFileOutputData(vetFileOutputData);
        
        return ssnVerification;
    }
    



    /**
	 * @param veteranAndSpouseData
	 * @return the list of output data.
	 */
    private List convertToOutputData(List veteranAndSpouseData)
    {
        List ssnVerificationDataList = new ArrayList();

        SSASSNVerificationData ssnVerificationData = null;

        if (veteranAndSpouseData != null)
        {
            for (Iterator iter = veteranAndSpouseData.iterator(); iter
                    .hasNext();)
            {
                Object[] dataRow = (Object[]) iter.next();
                try
                {
                    ssnVerificationData = new SSASSNVerificationData();

                    //SSN ID indicates that the SSN Verification is for a
                    // spouse/dependent
                    ssnVerificationData.setSsnId((String) dataRow[0]);
                    ssnVerificationData.setSsn((String) dataRow[1]);

                    String relationshipType = (String) dataRow[2];
                    //If Spouse present, set spouse code to 1
                    if (Relationship.CODE_SPOUSE.getCode().equals(
                            relationshipType))
                    {
                        ssnVerificationData.setSpouseCode("1");
                    }
                    ssnVerificationData.setLastName((String) dataRow[3]);
                    ssnVerificationData.setFirstName((String) dataRow[4]);
                    ssnVerificationData.setMiddleInitial((String) dataRow[5]);
                    ssnVerificationData
                            .setDateOfBirth((ImpreciseDate) dataRow[6]);
                    ssnVerificationData.setGender((String) dataRow[7]);
                    ssnVerificationData.setHECInternalID(generateControlIdentifier());
                    ssnVerificationData
                            .setSsaVerificationStatus(lookupService
                                    .getSSAVerificationStatusByCode(SSAVerificationStatus.IN_PROCESS
                                            .getCode()));
                } catch (ServiceException e)
                {
                    logger.error("Error while converting to output data for veteran and spouse data", e);
                }

                ssnVerificationDataList.add(ssnVerificationData);
            }
        }

        return ssnVerificationDataList;
    }
    


    /**
     * @param outputFileData the output file data
     * @return the ssn verification.
     */
    private List createSummaryStatisticsData(SSNVerification ssnVerification,
			List outputFileData) {

		// Add the count of this data to any existing count.
		ssnVerification.setCountSentToSsa(new Integer(ssnVerification
				.getCountSentToSsa().intValue()
				+ outputFileData.size()));

		ssnVerification.setSsaSentDate(Calendar.getInstance().getTime());

		// Create SSNVerificationDetail data
		List newVerificationDetails = new ArrayList();

		SSNVerificationDetail ssnVerificationDetail = null;
		for (Iterator iter = outputFileData.iterator(); iter.hasNext();) {
			
			//Exception should not happen at the batch level . Log and Continue.
			try {
				SSASSNVerificationData ssaSSNVerificationData = (SSASSNVerificationData) iter
						.next();

				ssnVerificationDetail = new SSNVerificationDetail();

				ssnVerificationDetail.setHECInternalId(ssaSSNVerificationData
						.getHECInternalID());

				// If ssn id is null, that means the data is for a vet. Get the
				// person Id by VPID(the HEC Internal ID)
				if (ssaSSNVerificationData.getSsnId() == null) {
					ssnVerificationDetail.setPersonId(new BigDecimal(
							ssaSSNVerificationData.getPersonId()));
				} else {
					ssnVerificationDetail.setSsnId(new BigDecimal(
							ssaSSNVerificationData.getSsnId()));
				}
				ssnVerificationDetail.setSsnText(ssaSSNVerificationData
						.getSsn());

				// Add the SSNVerifficationDetail to SSNVerification
				// ssnVerification.addSSNVerificationDetail(ssnVerificationDetail);
				ssnVerificationDetail.setSsnVerification(ssnVerification);
				newVerificationDetails.add(ssnVerificationDetail);
			} catch (Exception e) {
				   //Continue to the next object.
				   logger.error("createSummaryStatisticsData Exception:", e);
			}
		}
		return newVerificationDetails;
	}

    
    private void spawnThread(SSNVerificationProcessExecutionContext context,
    		Object[] data) {
        
    	if ( logger.isDebugEnabled())
        {
        	logger.debug(" RequestSSNVerification.spawnThread() taskCount = " + getTaskCount(context));
        }
		try {
			while (getTaskCount(context) >= throttleTaskCountThreshold) {
				Thread.sleep(spawnRetryPeriod);
			}
			initThreadCreatorAndIncrementTaskCount(context);
			ThreadPool threadPool = getThreadPool(context);
			// get a new Thread bean
			RequestSSNVerificationProcessSpawnedThreadTask task =
				(RequestSSNVerificationProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			
			task.setContext(context);
			task.setAuditInfo(getAuditInfo()); 
			task.setProcessName(this.getProcessName());
			task.setAcquiredData(data);	
			task.setSSNVerificationStatus(SSN_VERIFICATION_IN_PROCESS);
			task.setGenerator(generator);

			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("SSNVerificationProcessSpawnedThreadTask was interrupted while it was " +
						"spawning a thread. ", e); 
		}
    }

}