public boolean start(
      WorkflowSummary workflowSummary, Step step, StepRuntimeSummary runtimeSummary) {
    StepRuntime.Result result =
        getStepRuntime(runtimeSummary.getType())
            .start(workflowSummary, step, cloneSummary(runtimeSummary));
    runtimeSummary.mergeRuntimeUpdate(result.getTimeline(), result.getArtifacts());
    switch (result.getState()) {
      case CONTINUE:
      case DONE:
        runtimeSummary.markExecuting(tracingManager);
        return result.shouldPersist();
      case USER_ERROR:
        markTerminatedWithMetric(
            runtimeSummary, result.getState(), getUserErrorStatus(runtimeSummary));
        return false;
      case PLATFORM_ERROR:
        markTerminatedWithMetric(
            runtimeSummary, result.getState(), getPlatformErrorStatus(runtimeSummary));
        return false;
      case FATAL_ERROR:
        markTerminatedWithMetric(
            runtimeSummary, result.getState(), StepInstance.Status.FATALLY_FAILED);
        return false;
      case STOPPED:
        markTerminatedWithMetric(runtimeSummary, result.getState(), StepInstance.Status.STOPPED);
        return false;
      case TIMED_OUT:
        markTerminatedWithMetric(runtimeSummary, result.getState(), StepInstance.Status.TIMED_OUT);
        return false;
      default:
        throw new MaestroInternalError(
            "Entered an unexpected result state [%s] for step %s when starting",
            result.getState(), runtimeSummary.getIdentity());
    }
  }
@Test
  public void testStartFailure() {
    StepInstance.StepRetry stepRetry = new StepInstance.StepRetry();
    stepRetry.setErrorRetryLimit(1);
    stepRetry.setPlatformRetryLimit(1);
    stepRetry.incrementByStatus(StepInstance.Status.PLATFORM_FAILED);
    count.incrementAndGet();
    StepRuntimeSummary summary =
        StepRuntimeSummary.builder().type(StepType.NOOP).stepRetry(stepRetry).build();
    boolean ret = runtimeManager.start(workflowSummary, null, summary);
    assertFalse(ret);
    assertEquals(StepInstance.Status.USER_FAILED, summary.getRuntimeState().getStatus());
    assertNotNull(summary.getRuntimeState().getEndTime());
    assertNotNull(summary.getRuntimeState().getModifyTime());
    assertEquals(1, summary.getPendingRecords().size());
    assertEquals(
        StepInstance.Status.NOT_CREATED, summary.getPendingRecords().get(0).getOldStatus());
    assertEquals(
        StepInstance.Status.USER_FAILED, summary.getPendingRecords().get(0).getNewStatus());
    assertTrue(summary.getArtifacts().isEmpty());

    stepRetry.incrementByStatus(StepInstance.Status.USER_FAILED);
    ret = runtimeManager.start(workflowSummary, null, summary);
    assertFalse(ret);
    assertEquals(StepInstance.Status.FATALLY_FAILED, summary.getRuntimeState().getStatus());
    assertNotNull(summary.getRuntimeState().getEndTime());
    assertNotNull(summary.getRuntimeState().getModifyTime());
    assertEquals(2, summary.getPendingRecords().size());
    assertEquals(
        StepInstance.Status.USER_FAILED, summary.getPendingRecords().get(1).getOldStatus());
    assertEquals(
        StepInstance.Status.FATALLY_FAILED, summary.getPendingRecords().get(1).getNewStatus());
    assertTrue(summary.getArtifacts().isEmpty());
  }