org.springframework.boot.autoconfigure.jackson.Jackson2ObjectMapperBuilderCustomizer#org.springframework.batch.core.StepExecution源码实例Demo

@Test
public void givenReferenceOutput_whenStep1Executed_thenSuccess() throws Exception {

    // given
    FileSystemResource expectedResult = new FileSystemResource(EXPECTED_OUTPUT);
    FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT);

    // when
    JobExecution jobExecution = jobLauncherTestUtils.launchStep("step1", defaultJobParameters());
    Collection<StepExecution> actualStepExecutions = jobExecution.getStepExecutions();
    ExitStatus actualJobExitStatus = jobExecution.getExitStatus();

    // then
    assertThat(actualStepExecutions.size(), is(1));
    assertThat(actualJobExitStatus.getExitCode(), is("COMPLETED"));
    AssertFile.assertFileEquals(expectedResult, actualResult);
}
 

@Override
public StepExecution getStepExecution(Long jobExecutionId, Long stepExecutionId)
		throws NoSuchJobExecutionException, NoSuchStepExecutionException {
	JobExecution jobExecution = getJobExecution(jobExecutionId);
	StepExecution stepExecution = stepExecutionDao.getStepExecution(jobExecution, stepExecutionId);
	if (stepExecution == null) {
		throw new NoSuchStepExecutionException("There is no StepExecution with jobExecutionId=" + jobExecutionId
				+ " and id=" + stepExecutionId);
	}
	try {
		stepExecution.setExecutionContext(executionContextDao.getExecutionContext(stepExecution));
	}
	catch (Exception e) {
		logger.info("Cannot load execution context for step execution: " + stepExecution);
	}
	return stepExecution;
}
 

@Test
public void testSequentialTransitionAndSplit() {
	setupContextForGraph("AAA && FFF 'FAILED' -> EEE && <BBB||CCC> && DDD");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(5, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("BBB_0"));
	assertTrue(stepNames.contains("CCC_0"));
	assertTrue(stepNames.contains("DDD_0"));
	assertTrue(stepNames.contains("FFF_0"));
	List<StepExecution> sortedStepExecution =
			getSortedStepExecutions(stepExecutions);
	assertEquals("AAA_0", sortedStepExecution.get(0).getStepName());
	assertEquals("DDD_0", sortedStepExecution.get(4).getStepName());
}
 

@Override
public void sendEmail(AlarmChecker checker, int sequenceCount, StepExecution stepExecution) {
    List<String> receivers = userGroupService.selectEmailOfMember(checker.getuserGroupId());

    if (receivers.size() == 0) {
        return;
    }

    try{
        AlarmMailTemplate mailTemplate = new AlarmMailTemplate(checker, pinpointUrl, batchEnv, sequenceCount);
        MimeMessage message = springMailSender.createMimeMessage();
        message.setFrom(senderEmailAddress);
        message.setRecipients(Message.RecipientType.TO, getReceivers(receivers));

        final String subject =  mailTemplate.createSubject();
        message.setSubject(subject);
        message.setContent(mailTemplate.createBody(), "text/html");
        springMailSender.send(message);
        logger.info("send email : {}", subject);
    }catch(Exception e){
        logger.error("can't send alarm email. {}", checker.getRule(), e);
    }
}
 

private List<Table> buildTable(List<StepExecution> stepContexts) {
	return stepContexts.stream().map(step -> {
		String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
				.format(step.getStartTime());
		long millis = ChronoUnit.MILLIS.between(step.getStartTime().toInstant(),
				step.getEndTime().toInstant());
		ExecutionContext context = step.getExecutionContext();
		ExecutionResultReport entity = (ExecutionResultReport) context.get("entity");
		if (entity == null) {
			return null;
		}
		String projectName = TrainPostReleaseReleaserTask.class
				.isAssignableFrom(entity.getReleaserTaskType()) ? "postRelease"
						: entity.getProjectName();
		return new Table(date, time(millis), projectName, entity.getShortName(),
				entity.getDescription(), entity.getState(), entity.getExceptions());
	}).filter(Objects::nonNull).collect(Collectors.toCollection(LinkedList::new));
}
 

@Test
public void testSequentialAndSplit() {
	setupContextForGraph("AAA && <BBB||CCC||DDD> && EEE");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(5, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("BBB_0"));
	assertTrue(stepNames.contains("CCC_0"));
	assertTrue(stepNames.contains("DDD_0"));
	assertTrue(stepNames.contains("EEE_0"));
	List<StepExecution> sortedStepExecution =
			getSortedStepExecutions(stepExecutions);
	assertEquals("AAA_0", sortedStepExecution.get(0).getStepName());
	assertEquals("EEE_0", sortedStepExecution.get(4).getStepName());
}
 

/**
 * Constructor for the StepExecution to initialize the DTO.
 * @param original the StepExecution to build this DTO around.
 */
public JobExecutionEvent(JobExecution original) {
	this.jobParameters = new JobParametersEvent(
			original.getJobParameters().getParameters());
	this.jobInstance = new JobInstanceEvent(original.getJobInstance().getId(),
			original.getJobInstance().getJobName());
	for (StepExecution stepExecution : original.getStepExecutions()) {
		this.stepExecutions.add(new StepExecutionEvent(stepExecution));
	}
	this.status = original.getStatus();
	this.startTime = original.getStartTime();
	this.createTime = original.getCreateTime();
	this.endTime = original.getEndTime();
	this.lastUpdated = original.getLastUpdated();
	this.exitStatus = new ExitStatus(original.getExitStatus());
	this.executionContext = original.getExecutionContext();
	this.failureExceptions = original.getFailureExceptions();
	this.jobConfigurationName = original.getJobConfigurationName();
	this.setId(original.getId());
	this.setVersion(original.getVersion());
}
 

public static StepExecutionEventInfo transformToStepExecutionEventInfo(final StepExecution stepExecution, final String applicationName) {
    final StepExecutionEventInfo stepExecutionEventInfo = new StepExecutionEventInfo();
    stepExecutionEventInfo.setApplicationName(applicationName);

    stepExecutionEventInfo.setJobName(stepExecution.getJobExecution().getJobInstance().getJobName());
    stepExecutionEventInfo.setStepName(stepExecution.getStepName());
    stepExecutionEventInfo.setExitStatus(BatchToResourceMapper.map(stepExecution.getExitStatus()));

    stepExecutionEventInfo.setReadCount(stepExecution.getReadCount());
    stepExecutionEventInfo.setWriteCount(stepExecution.getWriteCount());
    stepExecutionEventInfo.setCommitCount(stepExecution.getCommitCount());
    stepExecutionEventInfo.setRollbackCount(stepExecution.getRollbackCount());
    stepExecutionEventInfo.setReadSkipCount(stepExecution.getReadSkipCount());
    stepExecutionEventInfo.setProcessSkipCount(stepExecution.getProcessSkipCount());
    stepExecutionEventInfo.setWriteSkipCount(stepExecution.getWriteSkipCount());
    return stepExecutionEventInfo;
}
 

private static long createSampleJob(String jobName, int jobExecutionCount) {
	JobInstance instance = jobRepository.createJobInstance(jobName, new JobParameters());
	jobInstances.add(instance);
	TaskExecution taskExecution = dao.createTaskExecution(jobName, new Date(), new ArrayList<>(), null);
	Map<String, JobParameter> jobParameterMap = new HashMap<>();
	jobParameterMap.put("foo", new JobParameter("FOO", true));
	jobParameterMap.put("bar", new JobParameter("BAR", false));
	JobParameters jobParameters = new JobParameters(jobParameterMap);
	JobExecution jobExecution;
	for (int i = 0; i < jobExecutionCount; i++) {
		jobExecution = jobRepository.createJobExecution(instance, jobParameters, null);
		taskBatchDao.saveRelationship(taskExecution, jobExecution);
		StepExecution stepExecution = new StepExecution("foobar", jobExecution);
		jobRepository.add(stepExecution);
	}
	return taskExecution.getExecutionId();
}
 

/**
 * Assert that by using the {@link ExecutionContextJacksonMixIn} Jackson renders the
 * Step Execution Context correctly.
 *
 * @throws JsonProcessingException if a Json generation error occurs.
 */
@Test
public void testSerializationOfSingleStepExecution() throws JsonProcessingException {

	final ObjectMapper objectMapper = new ObjectMapper();

	objectMapper.addMixIn(StepExecution.class, StepExecutionJacksonMixIn.class);
	objectMapper.addMixIn(ExecutionContext.class, ExecutionContextJacksonMixIn.class);

	final StepExecution stepExecution = getStepExecution();
	final String result = objectMapper.writeValueAsString(stepExecution);

	assertThat(result, not(containsString("\"executionContext\":{\"dirty\":true,\"empty\":false}")));
	assertThat(result, containsString("\"executionContext\":{\"dirty\":true,\"empty\":false,\"values\":[{"));

	assertThat(result, containsString("{\"counter\":1234}"));
	assertThat(result, containsString("{\"myDouble\":1.123456}"));
	assertThat(result, containsString("{\"Josh\":4444444444}"));
	assertThat(result, containsString("{\"awesomeString\":\"Yep\"}"));
	assertThat(result, containsString("{\"hello\":\"world\""));
	assertThat(result, containsString("{\"counter2\":9999}"));
}
 

/**
 * Generates an appropriate statistic of the processed, <br>
 * delegates the cleanup and the metric notification.
 * 
 * @param se
 *            the StepExecution
 */
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW)
public ExitStatus afterStep(StepExecution se) {
    LOG.info(se);

    statisticDao.save(createImportStatistic(se));

    if (CampusProcessStep.IMPORT_CONTROLFILE.name().equalsIgnoreCase(se.getStepName())) {
        if (se.getWriteCount() != getFixedNumberOfFilesToBeExported()) {
            // if (se.getReadCount() != getFixedNumberOfFilesToBeExported() || se.getWriteCount() != getFixedNumberOfFilesToBeExported()) {
            notifyMetrics(se);
            return ExitStatus.FAILED;
        }
    }

    removeOldDataIfExist(se);
    notifyMetrics(se);

    return null;
}
 

@Test
public void testDeserializationOfSingleJobExecution() throws IOException {

	final ObjectMapper objectMapper = DataFlowTemplate.prepareObjectMapper(new ObjectMapper());

	final InputStream inputStream = JobExecutionDeserializationTests.class
			.getResourceAsStream("/SingleJobExecutionJson.txt");

	final String json = new String(StreamUtils.copyToByteArray(inputStream));

	final JobExecutionResource jobExecutionInfoResource = objectMapper.readValue(json, JobExecutionResource.class);

	assertNotNull(jobExecutionInfoResource);
	assertEquals(Long.valueOf(1), jobExecutionInfoResource.getJobId());
	assertEquals("ff.job", jobExecutionInfoResource.getName());
	assertEquals("COMPLETED", jobExecutionInfoResource.getJobExecution().getStatus().name());
	assertEquals(1, jobExecutionInfoResource.getJobExecution().getStepExecutions().size());

	final StepExecution stepExecution = jobExecutionInfoResource.getJobExecution().getStepExecutions().iterator().next();
	assertNotNull(stepExecution);

	final ExecutionContext stepExecutionExecutionContext = stepExecution.getExecutionContext();

	assertNotNull(stepExecutionExecutionContext);
	assertEquals(2, stepExecutionExecutionContext.size());
}
 

@Test
public void readTest3() {
    StepExecution stepExecution = new StepExecution("alarmStep", null);
    ExecutionContext executionContext = new ExecutionContext();
    stepExecution.setExecutionContext(executionContext);
    
    AlarmServiceImpl alarmService = new AlarmServiceImpl(mock(AlarmDao.class)) {
        @Override
        public java.util.List<Rule> selectRuleByApplicationId(String applicationId) {
            return new LinkedList<>();
        }
    };
    
    AlarmReader reader = new AlarmReader(dataCollectorFactory, applicationIndexDao, alarmService);
    reader.beforeStep(stepExecution);
    assertNull(reader.read());
}
 

@Test
public void givenMockedStep_whenReaderCalled_thenSuccess() throws Exception {

    // given
    StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(defaultJobParameters());

    // when
    StepScopeTestUtils.doInStepScope(stepExecution, () -> {
        BookRecord bookRecord;
        itemReader.open(stepExecution.getExecutionContext());
        while ((bookRecord = itemReader.read()) != null) {

            // then
            assertThat(bookRecord.getBookName(), is("Foundation"));
            assertThat(bookRecord.getBookAuthor(), is("Asimov I."));
            assertThat(bookRecord.getBookISBN(), is("ISBN 12839"));
            assertThat(bookRecord.getBookFormat(), is("hardcover"));
            assertThat(bookRecord.getPublishingYear(), is("2018"));
        }
        itemReader.close();
        return null;
    });
}
 

@Override
public ExitStatus afterStep(StepExecution stepExecution) {
	// Calculate step execution time
	// Why is stepExecution.getEndTime().getTime() not available here? (see AbstractStep)
	long stepDuration = System.currentTimeMillis() - stepExecution.getStartTime().getTime();
	meterRegistry.gauge(METRIC_NAME, Arrays.asList(//
			new ImmutableTag("context", getStepExecutionIdentifier(stepExecution)), //
			new ImmutableTag("name", "duration")//
	), stepDuration);
	long itemCount = stepExecution.getWriteCount() + stepExecution.getSkipCount();
	meterRegistry.gauge(METRIC_NAME, Arrays.asList(//
			new ImmutableTag("context", getStepExecutionIdentifier(stepExecution)), //
			new ImmutableTag("name", "item.count")//
	), itemCount);
	// Calculate execution time per item
	long durationPerItem = 0;
	if (itemCount > 0) {
		durationPerItem = stepDuration / itemCount;
	}
	meterRegistry.gauge(METRIC_NAME, Arrays.asList(//
			new ImmutableTag("context", getStepExecutionIdentifier(stepExecution)), //
			new ImmutableTag("name", "item.duration")//
	), durationPerItem);
	// Export metrics from StepExecution to MetricRepositories
	Set<Entry<String, Object>> metrics = stepExecution.getExecutionContext().entrySet();
	for (Entry<String, Object> metric : metrics) {
		if (metric.getValue() instanceof Number) {
			meterRegistry.gauge(METRIC_NAME, Arrays.asList(//
					new ImmutableTag("context", getStepExecutionIdentifier(stepExecution)), //
					new ImmutableTag("name", metric.getKey())//
			), (Number) metric.getValue());
		}
	}
	return null;
}
 

/**
 * Prepares the metric data and delegates the actual notification to the {@link CampusNotifier}.
 * 
 * @param se
 *            the StepExecution
 */
private void notifyMetrics(StepExecution se) {
    if (CampusProcessStep.IMPORT_CONTROLFILE.name().equalsIgnoreCase(se.getStepName())) {
        campusNotifier.notifyStartOfImportProcess();
        CampusStatistics.EXPORT_STATUS exportStatus = CampusStatistics.EXPORT_STATUS.OK;

        if (se.getReadCount() != getFixedNumberOfFilesToBeExported() && se.getReadCount() != 2 * getFixedNumberOfFilesToBeExported()) {
            // THE CASE THAT THE EXPORT FILE (CONTROL FILE) HASN'T BEEN CREATED YET
            if (se.getReadCount() == 0) {
                exportStatus = CampusStatistics.EXPORT_STATUS.NO_EXPORT;
            }
            // THE CASE OF EXPORTING LESS THAN THE EXPECTED FILES (LESS THAN 8(ONLY CURRENT) OR LESS THAN 16 (CURRENT AND NEXT)
            else {
                exportStatus = CampusStatistics.EXPORT_STATUS.INCOMPLETE_EXPORT;
            }
        }
        // THE CASE OF EXPORTING THE OLD FILES
        if (se.getWriteCount() != getFixedNumberOfFilesToBeExported()) {
            exportStatus = CampusStatistics.EXPORT_STATUS.NO_EXPORT;
        }

        campusNotifier.notifyExportStatus(exportStatus);
    }

    campusNotifier.notifyStepExecution(se);

}
 

@Test
public void testAfterJobJobExecutionHasBatchStatusFailedWhenWeHaveStepsWithStatusNotCompleted() throws Exception {
    JobExecution jobExecution = new JobExecution(1L);

    StepExecution failedStepExecution = createStepExecution(jobExecution, FAILED);
    StepExecution completedStepExecution = createStepExecution(jobExecution, COMPLETED);
    StepExecution stoppedStepExecution = createStepExecution(jobExecution, STOPPED);

    jobExecution.addStepExecutions(asList(failedStepExecution, completedStepExecution, stoppedStepExecution));

    changeStatusOnFailedStepsJobExecListener.afterJob(jobExecution);

    assertThat(jobExecution.getStatus()).isEqualTo(BatchStatus.FAILED);
}
 

@Test
public void singleTest() {
	setupContextForGraph("AAA");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	assertEquals(1, stepExecutions.size());
	StepExecution stepExecution = stepExecutions.iterator().next();
	assertEquals("AAA_0", stepExecution.getStepName());
}
 

@Test
public void nestedSplit() {
	setupContextForGraph("<<AAA || BBB > && CCC || DDD>", "--splitThreadCorePoolSize=5");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(4, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("BBB_0"));
	assertTrue(stepNames.contains("CCC_0"));
	assertTrue(stepNames.contains("DDD_0"));
}
 

@Test
public void testSequentialTransitionAndSplitFailed() {
	setupContextForGraph("AAA && failedStep 'FAILED' -> EEE && FFF && <BBB||CCC> && DDD");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(3, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("failedStep_0"));
	assertTrue(stepNames.contains("EEE_0"));
}
 

@Test
public void testSequentialAndFailedSplit() {
	setupContextForGraph("AAA && <BBB||failedStep||DDD> && EEE");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(4, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("BBB_0"));
	assertTrue(stepNames.contains("DDD_0"));
	assertTrue(stepNames.contains("failedStep_0"));
}
 

@Override
public void beforeStep(StepExecution stepExecution) {
    ExecutionContext executionContext = stepExecution
      .getJobExecution()
      .getExecutionContext();
    this.lines = (List<Line>) executionContext.get("lines");
    fu = new FileUtils("output.csv");
    logger.debug("Lines Writer initialized.");
}
 

@Test
public void testNoPartitions() throws Exception {
	DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
			this.jobExplorer, this.resource, "step1");
	handler.setEnvironment(this.environment);

	StepExecution stepExecution = new StepExecution("step1", new JobExecution(1L));

	when(this.splitter.split(stepExecution, 1)).thenReturn(new HashSet<>());

	Collection<StepExecution> results = handler.handle(this.splitter, stepExecution);

	verify(this.taskLauncher, never()).launch((AppDeploymentRequest) any());
	assertThat(results.isEmpty()).isTrue();
}
 

@Test
public void testSuccessBasicTransition() {
	setupContextForGraph("AAA 'FAILED' -> BBB * -> CCC");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(2, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("CCC_0"));
}
 

/**
 * Constructor for the StepExecution to initialize the DTO.
 * @param stepExecution the StepExecution to build this DTO around.
 */
public StepExecutionEvent(StepExecution stepExecution) {
	super();
	Assert.notNull(stepExecution,
			"StepExecution must be provided to re-hydrate an existing StepExecutionEvent");
	Assert.notNull(stepExecution.getJobExecution(),
			"JobExecution must be provided to re-hydrate an existing StepExecutionEvent");
	setId(stepExecution.getId());
	this.jobExecutionId = stepExecution.getJobExecutionId();
	this.stepName = stepExecution.getStepName();

	this.status = stepExecution.getStatus();
	this.exitStatus = new ExitStatus(stepExecution.getExitStatus());
	this.executionContext = stepExecution.getExecutionContext();
	for (Throwable throwable : stepExecution.getFailureExceptions()) {
		this.failureExceptions.add(throwable);
	}
	this.terminateOnly = stepExecution.isTerminateOnly();

	this.endTime = stepExecution.getEndTime();
	this.lastUpdated = stepExecution.getLastUpdated();
	this.startTime = stepExecution.getStartTime();

	this.commitCount = stepExecution.getCommitCount();
	this.filterCount = stepExecution.getFilterCount();
	this.processSkipCount = stepExecution.getProcessSkipCount();
	this.readCount = stepExecution.getReadCount();
	this.readSkipCount = stepExecution.getReadSkipCount();
	this.rollbackCount = stepExecution.getRollbackCount();
	this.writeCount = stepExecution.getWriteCount();
	this.writeSkipCount = stepExecution.getWriteSkipCount();

}
 

private Collection<StepExecution> getStepExecutions() {
	JobExplorer jobExplorer = this.applicationContext.getBean(JobExplorer.class);
	List<JobInstance> jobInstances = jobExplorer.findJobInstancesByJobName("job", 0, 1);
	assertEquals(1, jobInstances.size());
	JobInstance jobInstance = jobInstances.get(0);
	List<JobExecution> jobExecutions = jobExplorer.getJobExecutions(jobInstance);
	assertEquals(1, jobExecutions.size());
	JobExecution jobExecution = jobExecutions.get(0);
	return jobExecution.getStepExecutions();
}
 

private Set<String> getStepNames(Collection<StepExecution> stepExecutions) {
	Set<String> result = new HashSet<>();
	for (StepExecution stepExecution : stepExecutions) {
		result.add(stepExecution.getStepName());
	}
	return result;
}
 

@Test
public void testSuccessBasicTransitionWithTransition() {
	setupContextForGraph("AAA 'FAILED' -> BBB && CCC 'FAILED' -> DDD '*' -> EEE");
	Collection<StepExecution> stepExecutions = getStepExecutions();
	Set<String> stepNames = getStepNames(stepExecutions);
	assertEquals(3, stepExecutions.size());
	assertTrue(stepNames.contains("AAA_0"));
	assertTrue(stepNames.contains("CCC_0"));
	assertTrue(stepNames.contains("EEE_0"));
	List<StepExecution> sortedStepExecution =
			getSortedStepExecutions(stepExecutions);
	assertEquals("AAA_0", sortedStepExecution.get(0).getStepName());
	assertEquals("EEE_0", sortedStepExecution.get(2).getStepName());
}
 

@Override
public boolean equals(Object obj) {

	if (!(obj instanceof StepExecution) || getId() == null) {
		return super.equals(obj);
	}
	StepExecution other = (StepExecution) obj;

	return this.stepName.equals(other.getStepName())
			&& (this.jobExecutionId == other.getJobExecutionId())
			&& getId().equals(other.getId());
}
 

@Test
public void testRuntimeException() throws Exception {
	StepExecution workerStep = new StepExecution("workerStep", new JobExecution(1L),
			2L);

	when(this.environment.containsProperty(
			DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID))
					.thenReturn(true);
	when(this.environment.containsProperty(
			DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID))
					.thenReturn(true);
	when(this.environment
			.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME))
					.thenReturn(true);
	when(this.environment
			.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME))
					.thenReturn("workerStep");
	when(this.beanFactory.getBeanNamesForType(Step.class))
			.thenReturn(new String[] { "workerStep", "foo", "bar" });
	when(this.environment.getProperty(
			DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID))
					.thenReturn("2");
	when(this.environment
			.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID))
					.thenReturn("1");
	when(this.jobExplorer.getStepExecution(1L, 2L)).thenReturn(workerStep);
	when(this.environment
			.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME))
					.thenReturn("workerStep");
	when(this.beanFactory.getBean("workerStep", Step.class)).thenReturn(this.step);
	doThrow(new RuntimeException("expected")).when(this.step).execute(workerStep);

	this.handler.run();

	verify(this.jobRepository).update(this.stepExecutionArgumentCaptor.capture());

	assertThat(this.stepExecutionArgumentCaptor.getValue().getStatus())
			.isEqualTo(BatchStatus.FAILED);
}