import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void checkpointAlgorithm() throws Exception {
	logger.info("starting checkpoint test");
	CountDownLatch latch = new CountDownLatch(1);
	JobOperator jobOperator = getJobOperator();
	Properties props = new Properties();
	long executionId = jobOperator.start(JOB_NAME, props);
	latch.await(10, SECONDS);

	JobInstance jobInstance = jobInstance(jobOperator.getJobInstance(executionId), JOB_NAME);
	JobExecution jobExecution = jobOperator.getJobExecutions(jobInstance).get(0);
	Properties jobProperties = jobExecution.getJobParameters();
	assertEquals("properties are: ", 0, jobProperties.size());
	assertEquals("batch failed because a NoRollbackException is throwed", FAILED, jobExecution.getBatchStatus());
	StepExecution stepExecution = jobOperator.getStepExecutions(executionId).get(0);
	PersistentCheckpointUserData persistentCheckpointUserData = (PersistentCheckpointUserData) stepExecution
			.getPersistentUserData();
	assertNull("persistent user data after step", persistentCheckpointUserData.getStartedAfterStep());
	Metric[] stepMetrics = stepExecution.getMetrics();
	stepMetrics(stepMetrics, 3, 10, 12, 2, 1);
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public String decide(StepExecution[] stepExecutions) throws Exception {
	if (stepExecutions.length != 1) {
		throw new IllegalStateException("Expecting stepExecutions array of size 1, found one of size = " + stepExecutions.length);
	}
	
	StepExecution stepExecution = stepExecutions[0];
	
       for (StepExecution stepExec : stepExecutions) {
           if (stepExec == null) {
               throw new Exception("Null StepExecution after flow.");
           }
           
           if (!stepExec.getBatchStatus().equals(BatchStatus.COMPLETED)) {
               throw new Exception("All step executions must be compelete before transitioning to a decider.");             
           }
           
       }
	
	// for our test 
	// <end exit-status="ThatsAllFolks" on="DECIDER_EXIT_STATUS*VERY GOOD INVOCATION" />
	return DECIDER_EXIT_STATUS + "*" + stepExecution.getExitStatus();
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public String decide(StepExecution[] stepExecutions) throws Exception {

    if (stepExecutions.length != 2) {
        throw new IllegalStateException("Expecting stepExecutions array of size 2, found one of size = " + stepExecutions.length);
    }

    for (StepExecution stepExec : stepExecutions) {
        if (stepExec == null) {
            throw new Exception("Null StepExecution after split.");
        }

        if (!stepExec.getBatchStatus().equals(BatchStatus.COMPLETED)) {
            throw new Exception("All step executions must be compelete before transitioning to a decider.");
        }

    }

    // <end exit-status="ThatsAllFolks" on="DECIDER_EXIT_STATUS*2" />
    return DECIDER_EXIT_STATUS + "*" + stepExecutions.length;
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public String decide(StepExecution[] stepExecutions) throws Exception {	
	if (stepExecutions.length != Integer.parseInt(numOfStepExecs)) {
		throw new Exception("Expecting stepExecutions array of size " +numOfStepExecs+ ", found one of size = " + stepExecutions.length);
	}
	
	/*we sort the stepExecutions to guarantee that the decider exit status, made up of the concatenated 
	step executions, will have the same order of step executions on subsequent job executions*/
	sortStepExecutionsByStepName(stepExecutions);
	
	/*
	 * exitStatus will be in the format:
	 *   "split1flow1step1:1235;split1flow1step2:1236;split1flow2step1:1237;"
	 * split the exitStatus at each ";" for each stepExecution
	 * split a stepExecution at ":" to get the stepName and stepExecutionId
	 */
	String exitStatus = "";
	for (StepExecution stepExecution: stepExecutions) {
		exitStatus += stepExecution.getStepName() + ":" + stepExecution.getStepExecutionId() + ";";
	}
	return exitStatus; //Decider return value should be set as the Job Exit Status
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
private void showStepState(StepExecution step) {
	Reporter.log("---------------------------<p>");
	Reporter.log("getStepName(): " + step.getStepName() + " - ");
	Reporter.log("getStepExecutionId(): " + step.getStepExecutionId() + " - ");
	Metric[] metrics = step.getMetrics();

	for (int i = 0; i < metrics.length; i++) {
		Reporter.log(metrics[i].getType() + ": " + metrics[i].getValue() + " - ");
	}

	Reporter.log("getStartTime(): " + step.getStartTime() + " - ");
	Reporter.log("getEndTime(): " + step.getEndTime() + " - ");
	Reporter.log("getBatchStatus(): " + step.getBatchStatus() + " - ");
	Reporter.log("getExitStatus(): " + step.getExitStatus()+"<p>");
	Reporter.log("---------------------------<p>");
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
private void showStepState(StepExecution step) {

        Reporter.log("---------------------------<p>");
        Reporter.log("getJobExecutionId(): " + step.getStepExecutionId() + " - ");

        Metric[] metrics = step.getMetrics();

        for (int i = 0; i < metrics.length; i++) {
            Reporter.log(metrics[i].getType() + ": " + metrics[i].getValue() + " - ");
        }

        Reporter.log("getStartTime(): " + step.getStartTime() + " - ");
        Reporter.log("getEndTime(): " + step.getEndTime() + " - ");
        Reporter.log("getBatchStatus(): " + step.getBatchStatus() + " - ");
        Reporter.log("getExitStatus(): " + step.getExitStatus());
        Reporter.log("---------------------------<p>");
    }
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
private void showStepState(StepExecution step) {


		Reporter.log("---------------------------");
		Reporter.log("getStepName(): " + step.getStepName() + " - ");
		Reporter.log("getJobExecutionId(): " + step.getStepExecutionId() + " - ");
		//System.out.print("getStepExecutionId(): " + step.getStepExecutionId() + " - ");
		Metric[] metrics = step.getMetrics();

		for (int i = 0; i < metrics.length; i++) {
			Reporter.log(metrics[i].getType() + ": " + metrics[i].getValue() + " - ");
		}

		Reporter.log("getStartTime(): " + step.getStartTime() + " - ");
		Reporter.log("getEndTime(): " + step.getEndTime() + " - ");
		//System.out.print("getLastUpdateTime(): " + step.getLastUpdateTime() + " - ");
		Reporter.log("getBatchStatus(): " + step.getBatchStatus() + " - ");
		Reporter.log("getExitStatus(): " + step.getExitStatus());
		Reporter.log("---------------------------");
	}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
/**
 * 
 * @param je  
 * @param stepName
 * @param numPartitionResults
 */
private void checkStepExecId(JobExecution je, String stepName, int numPartitionResults){
	List<StepExecution> stepExecs = jobOp.getStepExecutions(je.getExecutionId());
	
	Long stepExecId = null;
	for (StepExecution se : stepExecs) {
		if (se.getStepName().equals(stepName)) {
			stepExecId = se.getStepExecutionId();
			break;
		}
	}
	
	if (stepExecId == null) {
		throw new IllegalStateException("Didn't find step1 execution for job execution: " + je.getExecutionId());
	}
			
	String[] retvals = je.getExitStatus().split(",");
	assertWithMessage("Found different number of segments than expected in exit status string for job execution: " + je.getExecutionId(),
			 numPartitionResults, retvals.length);
	
	for(int i=0;i<retvals.length;i++){
		assertWithMessage("Did not return a number/numbers matching the stepExecId", stepExecId.longValue(), Long.parseLong(retvals[i]));
	}
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public List<StepExecution> getStepExecutions(long executionId)
		throws NoSuchJobExecutionException, JobSecurityException {

	logger.entering(sourceClass, "getStepExecutions", executionId);

	List<StepExecution> stepExecutions = new ArrayList<StepExecution>();

	IJobExecution jobEx = batchKernel.getJobExecution(executionId);
	if (jobEx == null){
		logger.fine("Job Execution: " + executionId + " not found");
		throw new NoSuchJobExecutionException("Job Execution: " + executionId + " not found");
	}
	if (isAuthorized(persistenceService.getJobInstanceIdByExecutionId(executionId))) {
		stepExecutions = persistenceService.getStepExecutionsForJobExecution(executionId);
	} else {
		logger.warning("getStepExecutions: The current user is not authorized to perform this operation");
		throw new JobSecurityException("The current user is not authorized to perform this operation");
	}

	logger.exiting(sourceClass, "getStepExecutions", stepExecutions);
	return stepExecutions;

}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void testEndNormallyExecuteBoth() throws Exception {
	JobOperator jo = BatchRuntime.getJobOperator();
	Properties props = new Properties();
	props.put("forceFailure", "true");
	long exec1Id = jo.start("nextOnStep1Failure", props);
	Thread.sleep(normalSleepTime);
	JobExecution je = jo.getJobExecution(exec1Id);
	Map<String, StepExecution> steps = new HashMap<String,StepExecution>();
	for (StepExecution se : jo.getStepExecutions(exec1Id)) { steps.put(se.getStepName(), se); }
	
	assertEquals("Job batch status", BatchStatus.COMPLETED, je.getBatchStatus());
	assertEquals("exit status", "COMPLETED", je.getExitStatus());
	assertEquals("Steps executed", 2, steps.size());
	assertEquals("Step failed", BatchStatus.FAILED, steps.get("step1").getBatchStatus());
	assertEquals("Step failed", BatchStatus.COMPLETED, steps.get("step2").getBatchStatus());
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void testCollectorPropertyResolver() throws Exception {
	long execId = jobOp.start("partitionPropertyResolverTest", null);
	Thread.sleep(sleepTime);
	JobExecution je = jobOp.getJobExecution(execId);

	assertEquals("batch status", BatchStatus.COMPLETED,je.getBatchStatus());

	List<StepExecution> stepExecutions = jobOp.getStepExecutions(execId);		
	String data = (String) stepExecutions.get(0).getPersistentUserData();
	String[] tokens = data.split(":");
	String[] dataItems = new String[tokens.length - 1]; // ignores before ':'
	for(int i=1; i < tokens.length; i++) {
		dataItems[i - 1] = tokens[i];
	}

	for(String s : dataItems) {
		String stepProp = s.substring(s.indexOf('#') + 1, s.indexOf('$'));
		String collectorProp = s.substring(s.indexOf('$') + 1);
		assertEquals("Step Property ", stepProp, collectorProp);
	}
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
@Ignore("Only works if you happen to have instance '6269' but we'll leave the logic to exercise getMostRecentStepExecutionsForJobInstance")
public void testStepExecution() throws Exception {
	IPersistenceManagerService ps = ServicesManagerImpl.getInstance().getPersistenceManagerService();
	Map<String, StepExecution> steps = ps.getMostRecentStepExecutionsForJobInstance(6269);

	String[] strings = new String[steps.size()];
	for (String name : steps.keySet().toArray(strings)) {
		StepExecutionImpl step = (StepExecutionImpl)steps.get(name);
		System.out.println("-------------------------------");
		System.out.println("SKSK: " + step.getStepName());
		System.out.println("SKSK: " + step.getStepExecutionId());
		System.out.println("SKSK: " + step.getJobExecutionId());
		System.out.println("SKSK: " + step.getExitStatus());
		System.out.println("SKSK: " + step.getStartTime());
		System.out.println("SKSK: " + step.getEndTime());
	}
	
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
@TCKCandidate("If we think this is well-specified enough")
public void testFailureAfterOnlyOneFlowFails() throws Exception {
	
	long theExecId = jobOp.start("splitFlowAggregateFailure", null);
	Thread.sleep(sleepTime);
	
	// Check job COMPLETED since some validation is crammed into the execution.
	JobExecution je = jobOp.getJobExecution(theExecId);
	assertEquals("Test failure", "FAILED", je.getBatchStatus().toString());

	List<StepExecution> stepExecutions = jobOp.getStepExecutions(theExecId);
	assertEquals("Number StepExecutions", 2, stepExecutions.size());

	for (StepExecution se : stepExecutions) {
		if (se.getStepName().equals("flow1step1")) {
			assertEquals("First flow should complete successfully", BatchStatus.COMPLETED, se.getBatchStatus());
		} else if (se.getStepName().equals("flow2step1")) {
			assertEquals("Second flow should fail", BatchStatus.FAILED, se.getBatchStatus());
		} else {
			throw new Exception("Unexpected step: " + se.getStepName());
		}
	}
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void testBatchDecision() throws Exception {
    JobOperator jobOperator = BatchRuntime.getJobOperator();
    Long executionId = jobOperator.start("myJob", new Properties());
    JobExecution jobExecution = jobOperator.getJobExecution(executionId);

    BatchTestHelper.keepTestAlive(jobExecution);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    List<String> executedSteps = new ArrayList<>();
    for (StepExecution stepExecution : stepExecutions) {
        executedSteps.add(stepExecution.getStepName());
    }

    assertEquals(2, stepExecutions.size());
    assertArrayEquals(new String[] {"step1", "step3"}, executedSteps.toArray());
    assertFalse(executedSteps.contains("step2"));
    assertEquals(BatchStatus.COMPLETED, jobExecution.getBatchStatus());
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void testBatchSplit() throws Exception {
    JobOperator jobOperator = BatchRuntime.getJobOperator();
    Long executionId = jobOperator.start("myJob", new Properties());
    JobExecution jobExecution = jobOperator.getJobExecution(executionId);

    BatchTestHelper.keepTestAlive(jobExecution);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    List<String> executedSteps = new ArrayList<>();
    for (StepExecution stepExecution : stepExecutions) {
        executedSteps.add(stepExecution.getStepName());
    }

    assertEquals(3, stepExecutions.size());
    assertTrue(executedSteps.contains("step1"));
    assertTrue(executedSteps.contains("step2"));
    assertTrue(executedSteps.contains("step3"));
    assertTrue(executedSteps.get(0).equals("step1") || executedSteps.get(0).equals("step2"));
    assertTrue(executedSteps.get(1).equals("step1") || executedSteps.get(1).equals("step2"));
    assertTrue(executedSteps.get(2).equals("step3"));
    assertEquals(BatchStatus.COMPLETED, jobExecution.getBatchStatus());
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Test
public void testRolledBackDuringWork() {
    final JobOperator jobOperator = BatchRuntime.getJobOperator();
    long executionId = jobOperator.start("txtest1", null);
    BatchStatus batchStatus = Batches.waitFor(jobOperator, executionId);
    Assert.assertEquals(batchStatus, BatchStatus.FAILED);
    Assert.assertEquals(TxErrorWriter1.written.intValue(), 3);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    Assert.assertEquals(stepExecutions.size(), 1);
    StepExecution stepExecution = stepExecutions.get(0);
    Metric[] metrics = stepExecution.getMetrics();
    assertMetric(Metric.MetricType.READ_COUNT, 2, metrics);
    assertMetric(Metric.MetricType.WRITE_COUNT, 2, metrics);
    assertMetric(Metric.MetricType.ROLLBACK_COUNT, 1, metrics);
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
private static Object[] asArray(final StepExecution n) {
    return new Object[] {
        n.getStepExecutionId(),
        n.getStepName(),
        n.getBatchStatus().name(),
        n.getExitStatus(),
        n.getStartTime() != null ? n.getStartTime().toString() : "",
        n.getEndTime() != null ? n.getEndTime().toString() : "",
        metric(n.getMetrics(), Metric.MetricType.READ_COUNT),
        metric(n.getMetrics(), Metric.MetricType.WRITE_COUNT),
        metric(n.getMetrics(), Metric.MetricType.COMMIT_COUNT),
        metric(n.getMetrics(), Metric.MetricType.ROLLBACK_COUNT),
        metric(n.getMetrics(), Metric.MetricType.READ_SKIP_COUNT),
        metric(n.getMetrics(), Metric.MetricType.PROCESS_SKIP_COUNT),
        metric(n.getMetrics(), Metric.MetricType.WRITE_SKIP_COUNT),
        metric(n.getMetrics(), Metric.MetricType.FILTER_COUNT)
    };
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public List<StepExecution> getStepExecutionsForJobExecution(final long execid) {
    final EntityManager em = emProvider.newEntityManager();
    try {
        final List<StepExecutionEntity> steps = em.createNamedQuery(StepExecutionEntity.Queries.FIND_BY_EXECUTION, StepExecutionEntity.class)
            .setParameter("executionId", execid)
            .getResultList();

        if (steps == null) {
            return Collections.emptyList();
        }

        final List<StepExecution> executions = new ArrayList<StepExecution>(steps.size());
        for (final StepExecutionEntity entity : steps) {
            final StepExecutionImpl stepEx = new StepExecutionImpl(execid, entity.getId());
            setStepExecutionData(entity, stepEx);
            executions.add(stepEx);
        }

        return executions;
    } finally {
        emProvider.release(em);
    }
}
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
private void updateStepExecution(final long stepExecutionId, final long jobExecId, final String batchStatus, final String exitStatus, final String stepName,
                                     final long readCount, final long writeCount, final long commitCount, final long rollbackCount,
                                     final long readSkipCount, final long processSkipCount, final long filterCount, final long writeSkipCount,
                                     final Timestamp startTime, final Timestamp endTime, final Serializable persistentData) {
//CHECKSTYLE:ON
        final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(jobExecId);
        if (executionInstanceData == null) {
            return;
        }

        synchronized (executionInstanceData.stepExecutions) {
            for (final StepExecution execution : executionInstanceData.stepExecutions) {
                if (execution.getStepExecutionId() == stepExecutionId) {
                    updateStepExecutionInstanceData(executionInstanceData, batchStatus, exitStatus, stepName,
                        readCount, writeCount, commitCount, rollbackCount,
                        readSkipCount, processSkipCount, filterCount, writeSkipCount,
                        startTime, endTime, persistentData,
                        data.stepExecutionInstanceData.get(stepExecutionId));
                    break;
                }
            }
        }
    }
 

import javax.batch.runtime.StepExecution; //导入依赖的package包/类
@Override
public void cleanUp(final long instanceId) {
    final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.remove(instanceId);
    if (jobInstanceData == null) {
        return;
    }

    synchronized (jobInstanceData.executions) {
        for (final Structures.ExecutionInstanceData executionInstanceData : jobInstanceData.executions) {
            data.executionInstanceData.remove(executionInstanceData.execution.getExecutionId());
            synchronized (executionInstanceData.stepExecutions) {
                for (final StepExecution stepExecution : executionInstanceData.stepExecutions) {
                    data.stepExecutionInstanceData.remove(stepExecution.getStepExecutionId());
                }
            }
        }
    }
    synchronized (jobInstanceData.checkpoints) {
        for (final CheckpointDataKey key : jobInstanceData.checkpoints) {
            data.checkpointData.remove(key);
        }
    }
}